queue overflow 30001: a send loop outruns one long code

"""Report Twilio senders whose queue is overflowing with 30001 or 21611.

Read only. GET requests and nothing else: give this an API Key with read access
rather than the account auth token. The repair is printed, never performed,
because this script holds a credential to an account that can send messages and
spend money.
"""
import argparse
import datetime as dt
import logging
import os
import sys

import requests

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("twilio_queue_overflow_audit")

HOST = "https://api.twilio.com"
BASE = HOST + "/2010-04-01"
MESSAGING = "https://messaging.twilio.com/v1"

# The same wall from two sides: 21611 rejects the request because the queue for
# that From is full, 30001 fails a message that got in and never drained.
OVERFLOW = (30001, 21611)
WAITING = ("queued", "accepted", "scheduled", "sending")


def error_code(message):
    """Read error_code as an integer, or None.

    It is null on every healthy message. Comparing the raw value against 30001
    is the mistake that reports a clean account the morning after an overflow.
    """
    raw = message.get("error_code")
    if raw is None or raw == "":
        return None
    try:
        return int(raw)
    except (TypeError, ValueError):
        return None


def queue_hours(segments, mps):
    """How many hours of sending a pile of segments represents. Pure.

    Segments, not messages: a three-segment body occupies three slots in the
    sender's queue.
    """
    rate = max(float(mps or 0), 0.01)
    return segments / (rate * 3600.0)


def tally(messages):
    """Bucket outbound messages by the sender that owns the queue. Pure.

    The key is `from`, because throughput and the queue behind it belong to the
    sending number. The Messaging Service is kept alongside, since that is what
    you would widen to fix it.
    """
    rows = {}
    for m in messages:
        if str(m.get("direction") or "").startswith("inbound"):
            continue
        key = m.get("from") or m.get("messaging_service_sid") or "unknown sender"
        row = rows.setdefault(key, {"total": 0, "overflow": 0, "queued": 0,
                                    "segments": 0, "service": None, "sids": []})
        row["total"] += 1
        try:
            row["segments"] += max(int(m.get("num_segments") or 1), 1)
        except (TypeError, ValueError):
            row["segments"] += 1
        if m.get("messaging_service_sid"):
            row["service"] = m.get("messaging_service_sid")
        if str(m.get("status") or "").lower() in WAITING:
            row["queued"] += 1
        if error_code(m) in OVERFLOW:
            row["overflow"] += 1
            if len(row["sids"]) < 3:
                row["sids"].append(m.get("sid"))
    return rows


def verdict(stats, mps=1.0, capacity_hours=10.0):
    """Classify one sender against what it can physically drain. Pure, so the
    throughput assumption is an argument rather than a hidden constant.

    Returns (state, detail).
    """
    total = int(stats.get("total") or 0)
    overflow = int(stats.get("overflow") or 0)
    waiting = int(stats.get("queued") or 0)
    segments = int(stats.get("segments") or 0) or total
    hours = queue_hours(segments, mps)
    tail = ("" if stats.get("service") else
            " Sent with a bare From, so there is one queue and no pool to spread "
            "it over.")

    if overflow:
        return ("overflow",
                "%d of %d rejected with 30001 or 21611. %d segment(s) is %.1f "
                "hours of sending at %.2f MPS, against a queue that holds about "
                "%.0f.%s" % (overflow, total, segments, hours, mps,
                             capacity_hours, tail))

    if hours >= capacity_hours:
        return ("over-capacity",
                "%d segment(s) is %.1f hours at %.2f MPS, past the roughly %.0f "
                "hour queue. Nothing failed yet, and the next run this size "
                "overflows.%s" % (segments, hours, mps, capacity_hours, tail))

    if hours >= capacity_hours / 2:
        return ("near-capacity",
                "%d segment(s) is %.1f hours at %.2f MPS against a queue of "
                "about %.0f. One retry storm, one duplicate batch or one "
                "template drifting into UCS-2 away from 30001.%s"
                % (segments, hours, mps, capacity_hours, tail))

    if waiting:
        return ("draining",
                "%d message(s) still queued or accepted; %d segment(s) is %.1f "
                "hours at %.2f MPS.%s" % (waiting, segments, hours, mps, tail))

    return ("clean", "%d message(s), %d segment(s), about %.1f hours at %.2f MPS"
            % (total, segments, hours, mps))


def get(session, url, **params):
    r = session.get(url, params=params, timeout=30)
    if r.status_code in (401, 403):
        raise SystemExit("%d from Twilio: check TWILIO_ACCOUNT_SID and that the "
                         "API key belongs to that account with read access"
                         % r.status_code)
    r.raise_for_status()
    return r.json()


def list_messages(session, account, since, limit):
    """Page Messages.json. There is no Status or ErrorCode filter here, so both
    error codes have to be found client-side."""
    url = "%s/Accounts/%s/Messages.json" % (BASE, account)
    params = {"PageSize": 1000, "DateSent>=": since}
    out = []
    while url and len(out) < limit:
        page = get(session, url, **params)
        out.extend(page.get("messages", []))
        nxt = page.get("next_page_uri")
        url, params = (HOST + nxt) if nxt else None, {}
    return out[:limit]


def pool_size(session, service_sid):
    """Count the senders in a Messaging Service pool. A service with one number
    has the throughput of one number."""
    url = "%s/Services/%s/PhoneNumbers" % (MESSAGING, service_sid)
    params = {"PageSize": 100}
    count = 0
    while url:
        page = get(session, url, **params)
        count += len(page.get("phone_numbers", []))
        url = (page.get("meta") or {}).get("next_page_url")
        params = {}
    return count


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--days", type=int, default=2,
                    help="how far back to read the Messages list")
    ap.add_argument("--max-messages", type=int, default=50000,
                    help="stop paging after this many messages")
    ap.add_argument("--mps", type=float, default=1.0,
                    help="segments per second for these senders: about 1 for a "
                         "US long code, higher for toll-free or a short code")
    ap.add_argument("--capacity-hours", type=float, default=10.0,
                    help="how many hours of segments the sender queue holds")
    args = ap.parse_args()

    account = os.environ.get("TWILIO_ACCOUNT_SID")
    key = os.environ.get("TWILIO_API_KEY")
    secret = os.environ.get("TWILIO_API_SECRET")
    if not (account and key and secret):
        log.error("set TWILIO_ACCOUNT_SID, TWILIO_API_KEY and TWILIO_API_SECRET "
                  "(an API Key with read access, not the auth token)")
        return 2

    session = requests.Session()
    session.auth = (key, secret)

    since = (dt.date.today() - dt.timedelta(days=args.days)).isoformat()
    messages = list_messages(session, account, since, args.max_messages)
    if not messages:
        log.info("no messages sent since %s", since)
        return 0

    senders = tally(messages)
    pools = {}
    bad = 0
    for sender, stats in sorted(senders.items()):
        state, detail = verdict(stats, args.mps, args.capacity_hours)
        line = "%-14s %s  %s" % (state, sender, detail)
        if state in ("clean", "draining"):
            log.info(line)
            continue
        bad += 1
        log.warning(line)
        if stats["sids"]:
            log.warning("  message sids: %s",
                        ", ".join(str(s) for s in stats["sids"]))
        service = stats.get("service")
        if service:
            if service not in pools:
                pools[service] = pool_size(session, service)
            log.warning("  %s has %d sender(s) in its pool: that is the "
                        "throughput you actually have.", service, pools[service])
            log.warning("  repair: POST %s/Services/%s/PhoneNumbers "
                        "PhoneNumberSid=PN... to widen the pool, and rate-limit "
                        "the producer to what the pool can drain.",
                        MESSAGING, service)
        else:
            log.warning("  repair: send through a Messaging Service "
                        "(MessagingServiceSid=MG...) instead of a bare From, add "
                        "senders to its pool, and rate-limit the producer. For "
                        "volume at this scale, toll-free or a short code.")

    log.info("%d sender(s) over %d day(s), %d over capacity",
             len(senders), args.days, bad)
    return 1 if bad else 0


if __name__ == "__main__":
    sys.exit(main())

/**
 * Report Twilio senders whose queue is overflowing with 30001 or 21611.
 *
 * Read only. GET requests and nothing else: give this an API Key with read
 * access rather than the account auth token. The repair is printed, never
 * performed.
 */
const HOST = 'https://api.twilio.com';
const BASE = `${HOST}/2010-04-01`;
const MESSAGING = 'https://messaging.twilio.com/v1';

// The same wall from two sides: 21611 rejects the request because the queue for
// that From is full, 30001 fails a message that got in and never drained.
const OVERFLOW = new Set([30001, 21611]);
const WAITING = new Set(['queued', 'accepted', 'scheduled', 'sending']);

/**
 * Read error_code as a number, or null. It is null on healthy messages, and
 * comparing the raw value is how the audit reports a clean account the morning
 * after an overflow.
 */
export function errorCode(message) {
  const raw = message.error_code;
  if (raw === null || raw === undefined || raw === '') return null;
  const n = Number(raw);
  return Number.isFinite(n) ? n : null;
}

/**
 * How many hours of sending a pile of segments represents. Pure. Segments, not
 * messages: a three-segment body occupies three slots in the queue.
 */
export function queueHours(segments, mps) {
  const rate = Math.max(Number(mps) || 0, 0.01);
  return segments / (rate * 3600);
}

/**
 * Bucket outbound messages by the sender that owns the queue. Pure. The key is
 * `from`, because throughput belongs to the sending number.
 */
export function tally(messages) {
  const rows = new Map();
  for (const m of messages) {
    if (String(m.direction ?? '').startsWith('inbound')) continue;
    const key = m.from || m.messaging_service_sid || 'unknown sender';
    if (!rows.has(key)) {
      rows.set(key, { total: 0, overflow: 0, queued: 0, segments: 0,
                      service: null, sids: [] });
    }
    const row = rows.get(key);
    row.total += 1;
    row.segments += Math.max(Number(m.num_segments ?? 1) || 1, 1);
    if (m.messaging_service_sid) row.service = m.messaging_service_sid;
    if (WAITING.has(String(m.status ?? '').toLowerCase())) row.queued += 1;
    if (OVERFLOW.has(errorCode(m))) {
      row.overflow += 1;
      if (row.sids.length < 3) row.sids.push(m.sid);
    }
  }
  return rows;
}

/**
 * Classify one sender against what it can physically drain. Pure, so the
 * throughput assumption is an argument. Returns [state, detail].
 */
export function verdict(stats, mps = 1.0, capacityHours = 10.0) {
  const total = Number(stats.total ?? 0);
  const overflow = Number(stats.overflow ?? 0);
  const waiting = Number(stats.queued ?? 0);
  const segments = Number(stats.segments ?? 0) || total;
  const hours = queueHours(segments, mps);
  const h = hours.toFixed(1);
  const rate = Number(mps).toFixed(2);
  const cap = capacityHours.toFixed(0);
  const tail = stats.service ? ''
    : ' Sent with a bare From, so there is one queue and no pool to spread it over.';

  if (overflow) {
    return ['overflow',
      `${overflow} of ${total} rejected with 30001 or 21611. ${segments} ` +
      `segment(s) is ${h} hours of sending at ${rate} MPS, against a queue ` +
      `that holds about ${cap}.${tail}`];
  }

  if (hours >= capacityHours) {
    return ['over-capacity',
      `${segments} segment(s) is ${h} hours at ${rate} MPS, past the roughly ` +
      `${cap} hour queue. Nothing failed yet, and the next run this size ` +
      `overflows.${tail}`];
  }

  if (hours >= capacityHours / 2) {
    return ['near-capacity',
      `${segments} segment(s) is ${h} hours at ${rate} MPS against a queue of ` +
      `about ${cap}. One retry storm, one duplicate batch or one template ` +
      `drifting into UCS-2 away from 30001.${tail}`];
  }

  if (waiting) {
    return ['draining',
      `${waiting} message(s) still queued or accepted; ${segments} segment(s) ` +
      `is ${h} hours at ${rate} MPS.${tail}`];
  }

  return ['clean',
    `${total} message(s), ${segments} segment(s), about ${h} hours at ${rate} MPS`];
}

function authHeader(key, secret) {
  return `Basic ${Buffer.from(`${key}:${secret}`).toString('base64')}`;
}

async function get(auth, url, params = {}) {
  const u = new URL(url);
  for (const [k, v] of Object.entries(params)) u.searchParams.set(k, v);
  const res = await fetch(u, { headers: { Authorization: auth } });
  if (res.status === 401 || res.status === 403) {
    throw new Error(`${res.status} from Twilio: check TWILIO_ACCOUNT_SID and ` +
                    'that the API key belongs to that account with read access');
  }
  if (!res.ok) throw new Error(`${res.status} from ${u.pathname}`);
  return res.json();
}

async function listMessages(auth, account, since, limit) {
  let url = `${BASE}/Accounts/${account}/Messages.json`;
  let params = { PageSize: 1000, 'DateSent>=': since };
  const out = [];
  while (url && out.length < limit) {
    const page = await get(auth, url, params);
    out.push(...(page.messages ?? []));
    url = page.next_page_uri ? HOST + page.next_page_uri : null;
    params = {};
  }
  return out.slice(0, limit);
}

async function poolSize(auth, serviceSid) {
  let url = `${MESSAGING}/Services/${serviceSid}/PhoneNumbers`;
  let params = { PageSize: 100 };
  let count = 0;
  while (url) {
    const page = await get(auth, url, params);
    count += (page.phone_numbers ?? []).length;
    url = page.meta?.next_page_url ?? null;
    params = {};
  }
  return count;
}

function flag(name, fallback) {
  const i = process.argv.indexOf(name);
  return i === -1 ? fallback : Number(process.argv[i + 1]);
}

async function main() {
  const account = process.env.TWILIO_ACCOUNT_SID;
  const key = process.env.TWILIO_API_KEY;
  const secret = process.env.TWILIO_API_SECRET;
  if (!account || !key || !secret) {
    console.error('set TWILIO_ACCOUNT_SID, TWILIO_API_KEY and TWILIO_API_SECRET ' +
                  '(an API Key with read access, not the auth token)');
    process.exitCode = 2;
    return;
  }
  const auth = authHeader(key, secret);
  const days = flag('--days', 2);
  const mps = flag('--mps', 1.0);
  const capacityHours = flag('--capacity-hours', 10.0);
  const since = new Date(Date.now() - days * 86400000).toISOString().slice(0, 10);

  const messages = await listMessages(auth, account, since, flag('--max-messages', 50000));
  if (messages.length === 0) {
    console.log(`no messages sent since ${since}`);
    return;
  }

  const senders = tally(messages);
  const pools = new Map();
  let bad = 0;
  for (const sender of [...senders.keys()].sort()) {
    const stats = senders.get(sender);
    const [state, detail] = verdict(stats, mps, capacityHours);
    const line = `${state.padEnd(14)} ${sender}  ${detail}`;
    if (state === 'clean' || state === 'draining') { console.log(line); continue; }
    bad += 1;
    console.warn(line);
    if (stats.sids.length) console.warn(`  message sids: ${stats.sids.join(', ')}`);
    if (stats.service) {
      if (!pools.has(stats.service)) {
        pools.set(stats.service, await poolSize(auth, stats.service));
      }
      console.warn(`  ${stats.service} has ${pools.get(stats.service)} sender(s) ` +
                   'in its pool: that is the throughput you actually have.');
      console.warn(`  repair: POST ${MESSAGING}/Services/${stats.service}` +
                   '/PhoneNumbers PhoneNumberSid=PN... to widen the pool, and ' +
                   'rate-limit the producer to what the pool can drain.');
    } else {
      console.warn('  repair: send through a Messaging Service ' +
                   '(MessagingServiceSid=MG...) instead of a bare From, add ' +
                   'senders to its pool, and rate-limit the producer. For volume ' +
                   'at this scale, toll-free or a short code.');
    }
  }

  console.log(`${senders.size} sender(s) over ${days} day(s), ${bad} over capacity`);
  process.exitCode = bad ? 1 : 0;
}

// Only run when invoked directly, so importing this module in the tests does not
// run main(), fail on the missing credentials and set a non-zero exit code.
if (import.meta.url === `file://${process.argv[1]}`) {
  main().catch((err) => { console.error(err.message); process.exitCode = 2; });
}

from twilio_queue_overflow_audit import queue_hours, tally, verdict


def sent(sid, sender, **extra):
    row = {"sid": sid, "from": sender, "status": "delivered", "num_segments": 1}
    row.update(extra)
    return row


def test_ten_hours_is_thirty_six_thousand_segments_at_one_mps():
    assert queue_hours(36000, 1) == 10.0
    assert round(queue_hours(3600, 0.5), 1) == 2.0


def test_a_zero_rate_does_not_divide_by_zero():
    assert queue_hours(100, 0) > 0


def test_tally_groups_by_sending_number_and_counts_segments():
    rows = tally([
        sent("SM1", "+15550001111", num_segments="3"),
        sent("SM2", "+15550001111", status="queued"),
        sent("SM3", "+15550002222", messaging_service_sid="MG1"),
        {"sid": "SM4", "from": "+15550001111", "direction": "inbound"},
    ])
    assert sorted(rows) == ["+15550001111", "+15550002222"]
    assert rows["+15550001111"]["segments"] == 4
    assert rows["+15550001111"]["queued"] == 1
    assert rows["+15550001111"]["service"] is None
    assert rows["+15550002222"]["service"] == "MG1"


def test_both_error_codes_count_as_the_same_wall():
    rows = tally([
        sent("SM1", "+1555", error_code=30001, status="failed"),
        sent("SM2", "+1555", error_code="21611", status="failed"),
        sent("SM3", "+1555"),
    ])
    assert rows["+1555"]["overflow"] == 2
    assert rows["+1555"]["sids"] == ["SM1", "SM2"]


def test_overflow_errors_are_the_headline():
    state, detail = verdict({"total": 40000, "overflow": 6000, "segments": 40000,
                             "service": "MG1"})
    assert state == "overflow"
    assert "11.1 hours" in detail


def test_a_sender_past_the_queue_depth_is_flagged_before_it_fails():
    state, detail = verdict({"total": 40000, "segments": 40000, "service": "MG1"})
    assert state == "over-capacity"
    assert "Nothing failed yet" in detail


def test_half_the_queue_is_already_worth_saying():
    state, detail = verdict({"total": 20000, "segments": 20000, "service": "MG1"})
    assert state == "near-capacity"
    assert "UCS-2" in detail


def test_a_bare_from_says_so():
    _, detail = verdict({"total": 40000, "segments": 40000})
    assert "bare From" in detail


def test_messages_still_waiting_are_draining_not_broken():
    state, detail = verdict({"total": 900, "segments": 900, "queued": 40,
                             "service": "MG1"})
    assert state == "draining"
    assert "40 message(s)" in detail


def test_a_small_run_is_clean():
    state, detail = verdict({"total": 100, "segments": 100, "service": "MG1"})
    assert state == "clean"
    assert "100 segment(s)" in detail


def test_three_segment_bodies_fill_the_queue_three_times_faster():
    # The same 18,000 messages, one segment each and then three each.
    assert verdict({"total": 18000, "segments": 18000, "service": "MG1"})[0] == "near-capacity"
    assert verdict({"total": 18000, "segments": 54000, "service": "MG1"})[0] == "over-capacity"

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { queueHours, tally, verdict } from './twilio-queue-overflow-audit.mjs';

const sent = (sid, from, extra = {}) => ({
  sid, from, status: 'delivered', num_segments: 1, ...extra,
});

test('ten hours is thirty six thousand segments at one MPS', () => {
  assert.equal(queueHours(36000, 1), 10);
  assert.equal(Number(queueHours(3600, 0.5).toFixed(1)), 2);
});

test('a zero rate does not divide by zero', () => {
  assert.ok(Number.isFinite(queueHours(100, 0)));
});

test('tally groups by sending number and counts segments', () => {
  const rows = tally([
    sent('SM1', '+15550001111', { num_segments: '3' }),
    sent('SM2', '+15550001111', { status: 'queued' }),
    sent('SM3', '+15550002222', { messaging_service_sid: 'MG1' }),
    { sid: 'SM4', from: '+15550001111', direction: 'inbound' },
  ]);
  assert.deepEqual([...rows.keys()].sort(), ['+15550001111', '+15550002222']);
  assert.equal(rows.get('+15550001111').segments, 4);
  assert.equal(rows.get('+15550001111').queued, 1);
  assert.equal(rows.get('+15550001111').service, null);
  assert.equal(rows.get('+15550002222').service, 'MG1');
});

test('both error codes count as the same wall', () => {
  const rows = tally([
    sent('SM1', '+1555', { error_code: 30001, status: 'failed' }),
    sent('SM2', '+1555', { error_code: '21611', status: 'failed' }),
    sent('SM3', '+1555'),
  ]);
  assert.equal(rows.get('+1555').overflow, 2);
  assert.deepEqual(rows.get('+1555').sids, ['SM1', 'SM2']);
});

test('overflow errors are the headline', () => {
  const [state, detail] = verdict({ total: 40000, overflow: 6000, segments: 40000,
                                    service: 'MG1' });
  assert.equal(state, 'overflow');
  assert.match(detail, /11\.1 hours/);
});

test('a sender past the queue depth is flagged before it fails', () => {
  const [state, detail] = verdict({ total: 40000, segments: 40000, service: 'MG1' });
  assert.equal(state, 'over-capacity');
  assert.match(detail, /Nothing failed yet/);
});

test('half the queue is already worth saying', () => {
  const [state, detail] = verdict({ total: 20000, segments: 20000, service: 'MG1' });
  assert.equal(state, 'near-capacity');
  assert.match(detail, /UCS-2/);
});

test('a bare From says so', () => {
  const [, detail] = verdict({ total: 40000, segments: 40000 });
  assert.match(detail, /bare From/);
});

test('messages still waiting are draining, not broken', () => {
  const [state, detail] = verdict({ total: 900, segments: 900, queued: 40,
                                    service: 'MG1' });
  assert.equal(state, 'draining');
  assert.match(detail, /40 message\(s\)/);
});

test('a small run is clean', () => {
  const [state, detail] = verdict({ total: 100, segments: 100, service: 'MG1' });
  assert.equal(state, 'clean');
  assert.match(detail, /100 segment\(s\)/);
});

test('three segment bodies fill the queue three times faster', () => {
  assert.equal(verdict({ total: 18000, segments: 18000, service: 'MG1' })[0],
               'near-capacity');
  assert.equal(verdict({ total: 18000, segments: 54000, service: 'MG1' })[0],
               'over-capacity');
});