feat(engagement): retention — three sweeps and one recorded refusal
ENGAGEMENT.md Phase 14, the last phase of the workstream. Four engagement
tables grew on every fire and nothing had ever deleted from any of them.
Three of them now have a horizon, swept nightly by one worker
(utils/engagementRetentionPrune.js — setInterval + unref + stop(), batched
1000 x 50, each table's failure caught on its own so a lock timeout on one
does not leave the other two unbounded):
engagement_sends 180 days engagement_sends_retain_days (7-3650)
engagement_cooldowns 30 days engagement_cooldowns_retain_days (2-3650)
engagement_outbox 30 days engagement_outbox_retain_days (2-3650)
The fourth, engagement_suppressions, does not expire, and that is the
recorded decision rather than an omission: a suppression is a standing
decision, and ageing out a hard bounce re-mails an address that already
bounced. The way out stays deliberate, and is now reachable per row.
Six decisions were settled by the org lead before any code. Two of them
widened the phase past what was offered:
* the send-log horizon is admin-configurable, so retention got a SCREEN
(Admin -> Engagement -> Retention) where team_activity and
user_notifications keep theirs in invisible settings rows. The send-log
horizon changes what an operator-facing page is able to show, so it has
to be visible; the other two came with it, because "what does this
deployment keep" is one question.
* the suppression purge, which cost a Phase 9 decision. The list
deliberately stripped address_hash from every row, so the only way out
was a window.prompt asking the operator to retype an address the screen
has never shown them. The row had no handle at all. The hash is now
returned: this route is admin-only and an admin can already suppress and
unsuppress any address they can name, so it grants no capability they
lack. GET /sends still strips its own.
The outbox sweep is TERMINAL-ONLY and that is a correctness rule: a
scheduled row is a send this deployment still intends to make (delay_seconds
can put one a day out) and a sending row may be mid-flight.
One shipped defect had to be fixed for the sweep to be a bound at all.
reclaimStale returned every stale sending row to scheduled, and MAX_ATTEMPTS
is consulted only on a graceful retry outcome — so a send that killed the
process mid-flight cycled sending -> scheduled -> sending forever, never
terminal, therefore never eligible for any sweep. It now fails an exhausted
row BEFORE reclaiming the rest; the order is the fix.
Two indexes (idx_engo_sweep, idx_engs_sweep): every existing index on those
tables has created_at in second position, which serves a per-rule window and
is useless to a whole-table horizon.
Proved twice: engagementRetentionSql.test.js against a real MariaDB (7
tests, incl. the acceptance case and the wrong reclaim order run
deliberately), and the live stack, where a 90-day-old cancelled row was
swept and a 90-day-old scheduled row survived.
Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
234
server/test/engagementRetention.test.js
Normal file
234
server/test/engagementRetention.test.js
Normal file
@@ -0,0 +1,234 @@
|
||||
// ── Retention: the engagement schema's sweep (ENGAGEMENT.md Phase 14) ──────
|
||||
//
|
||||
// The phase's acceptance line is that every one of the four tables has a stated
|
||||
// policy — a sweep with a horizon, or a recorded decision that it does not
|
||||
// expire — and that the cooldown horizon is CHECKED against the longest enabled
|
||||
// rule rather than picked. Both are pinned here, plus the three things building
|
||||
// it showed are silent when wrong:
|
||||
//
|
||||
// • **the outbox sweep is terminal-only.** A `scheduled` row is a message this
|
||||
// deployment still intends to send; `delay_seconds` can legitimately put one
|
||||
// a day out. Sweeping by age alone would cancel sends nobody cancelled, and
|
||||
// the operator would see only that the mail never arrived.
|
||||
// • **`reclaimStale` has to give up.** `MAX_ATTEMPTS` is consulted only on a
|
||||
// graceful `retry` outcome, so before Phase 14 a send that killed the
|
||||
// process mid-flight cycled sending → scheduled → sending forever, never
|
||||
// reached a terminal status, and was therefore never eligible for ANY
|
||||
// retention sweep. The bound depends on this.
|
||||
// • **an unreadable setting means the default, not an exception.** The sweep
|
||||
// runs on a timer with nobody watching.
|
||||
//
|
||||
// Point the DB at a closed port before requiring anything: the registries reach
|
||||
// utils/discordAnnounce, which builds the pool at require time.
|
||||
process.env.DB_HOST = '127.0.0.1'
|
||||
process.env.DB_PORT = '59999'
|
||||
|
||||
const { test, beforeEach, afterEach, after } = require('node:test')
|
||||
const assert = require('node:assert/strict')
|
||||
|
||||
const retention = require('../src/model/engagement/engagementRetention.model')
|
||||
const prune = require('../src/utils/engagementRetentionPrune')
|
||||
const cooldownsDb = require('../src/model/engagement/engagementCooldowns.db')
|
||||
const outboxDb = require('../src/model/engagement/engagementOutbox.db')
|
||||
const sendsDb = require('../src/model/engagement/engagementSends.db')
|
||||
const settings = require('../src/model/settings/settings.model')
|
||||
const rulesDb = require('../src/model/engagement/engagementRules.db')
|
||||
const db = require('../src/utils/db')
|
||||
|
||||
after(() => db.close())
|
||||
|
||||
const saved = new Map()
|
||||
function patch(mod, name, fn) {
|
||||
if (!saved.has(mod)) saved.set(mod, new Map())
|
||||
if (!saved.get(mod).has(name)) saved.get(mod).set(name, mod[name])
|
||||
mod[name] = fn
|
||||
}
|
||||
function restore() {
|
||||
for (const [mod, names] of saved) for (const [name, fn] of names) mod[name] = fn
|
||||
saved.clear()
|
||||
}
|
||||
|
||||
let world
|
||||
|
||||
beforeEach(() => {
|
||||
world = {
|
||||
settings: new Map(),
|
||||
longestCooldown: 0,
|
||||
// Every prune call, so a test can assert on the horizon it was given rather
|
||||
// than only on the fact that something was deleted.
|
||||
calls: { cooldowns: [], outbox: [], sends: [] },
|
||||
deleted: { cooldowns: 0, outbox: 0, sends: 0 },
|
||||
}
|
||||
patch(settings, 'get', async (key) => {
|
||||
if (!world.settings.has(key)) return null
|
||||
return world.settings.get(key)
|
||||
})
|
||||
patch(settings, 'set', async (key, value) => { world.settings.set(key, value) })
|
||||
patch(rulesDb, 'maxEnabledCooldownSeconds', async () => world.longestCooldown)
|
||||
patch(cooldownsDb, 'prune', async (before, limit) => {
|
||||
world.calls.cooldowns.push({ before, limit })
|
||||
return world.deleted.cooldowns
|
||||
})
|
||||
patch(outboxDb, 'pruneTerminal', async (before, limit) => {
|
||||
world.calls.outbox.push({ before, limit })
|
||||
return world.deleted.outbox
|
||||
})
|
||||
patch(sendsDb, 'prune', async (before, limit) => {
|
||||
world.calls.sends.push({ before, limit })
|
||||
return world.deleted.sends
|
||||
})
|
||||
})
|
||||
|
||||
afterEach(() => restore())
|
||||
|
||||
const daysBetween = (now, before) => Math.round((now.getTime() - before.getTime()) / 86400000)
|
||||
|
||||
// ── The policy ─────────────────────────────────────────────────────────────
|
||||
|
||||
test('defaults are the shipped policy when nothing is stored', async () => {
|
||||
const policy = await retention.get()
|
||||
assert.deepEqual(policy, { sends: 180, cooldowns: 30, outbox: 30 })
|
||||
})
|
||||
|
||||
test('a stored value is used, and written back as a string', async () => {
|
||||
await retention.set({ sends: 365 }, 7)
|
||||
assert.equal(world.settings.get('engagement_sends_retain_days'), '365')
|
||||
assert.equal((await retention.get()).sends, 365)
|
||||
})
|
||||
|
||||
test('the PUT is sparse — an absent horizon is left alone', async () => {
|
||||
await retention.set({ sends: 90 })
|
||||
await retention.set({ cooldowns: 45 })
|
||||
const policy = await retention.get()
|
||||
assert.equal(policy.sends, 90, 'the earlier write survives the second call')
|
||||
assert.equal(policy.cooldowns, 45)
|
||||
assert.equal(policy.outbox, 30, 'and the untouched one is still the default')
|
||||
})
|
||||
|
||||
test('out of range is refused rather than clamped', async () => {
|
||||
// Clamping would leave the screen describing a policy the deployment is not
|
||||
// running, which is worse than a visible error.
|
||||
await assert.rejects(() => retention.set({ cooldowns: 1 }), /between 2 and 3650/)
|
||||
await assert.rejects(() => retention.set({ sends: 5 }), /between 7 and 3650/)
|
||||
await assert.rejects(() => retention.set({ outbox: 99999 }), /between 2 and 3650/)
|
||||
await assert.rejects(() => retention.set({ sends: 12.5 }), /whole number/)
|
||||
assert.equal(world.settings.size, 0, 'nothing was written')
|
||||
})
|
||||
|
||||
test('an unknown key in the body is ignored, not rejected', async () => {
|
||||
// So a client that posts the whole object back is not coupled to the list.
|
||||
await retention.set({ sends: 200, suppressions: 30, nonsense: 1 })
|
||||
assert.equal((await retention.get()).sends, 200)
|
||||
assert.equal(world.settings.has('engagement_suppressions_retain_days'), false)
|
||||
})
|
||||
|
||||
test('a stored value outside the bounds falls back to the default', async () => {
|
||||
// A row written before the bounds existed, or by hand.
|
||||
world.settings.set('engagement_cooldowns_retain_days', '0')
|
||||
assert.equal((await retention.get()).cooldowns, 30)
|
||||
world.settings.set('engagement_cooldowns_retain_days', 'soon')
|
||||
assert.equal((await retention.get()).cooldowns, 30)
|
||||
})
|
||||
|
||||
test('an unreadable settings table yields defaults rather than throwing', async () => {
|
||||
patch(settings, 'get', async () => { throw new Error('pool is dead') })
|
||||
assert.deepEqual(await retention.get(), { sends: 180, cooldowns: 30, outbox: 30 })
|
||||
})
|
||||
|
||||
// ── The cooldown guard ─────────────────────────────────────────────────────
|
||||
|
||||
test('the cooldown horizon is checked against the longest ENABLED rule', async () => {
|
||||
world.longestCooldown = 86400 // the validated maximum, one day
|
||||
const ok = await retention.checkCooldownHorizon(30)
|
||||
assert.equal(ok.ok, true)
|
||||
assert.equal(ok.message, null)
|
||||
|
||||
// A horizon shorter than a live cooldown means a pruned row makes the next
|
||||
// fire a FIRST fire — the rule sends twice.
|
||||
const bad = await retention.checkCooldownHorizon(0.5)
|
||||
assert.equal(bad.ok, false)
|
||||
assert.match(bad.message, /can send twice/)
|
||||
assert.equal(bad.longestCooldownSeconds, 86400)
|
||||
})
|
||||
|
||||
test('the horizon equal to the longest cooldown is refused, not accepted', async () => {
|
||||
// Equality is the boundary where a row is pruned exactly as it stops being in
|
||||
// force; the check has to be <=, not <.
|
||||
world.longestCooldown = 2 * 86400
|
||||
assert.equal((await retention.checkCooldownHorizon(2)).ok, false)
|
||||
assert.equal((await retention.checkCooldownHorizon(3)).ok, true)
|
||||
})
|
||||
|
||||
test('no enabled rule means no warning at any horizon', async () => {
|
||||
world.longestCooldown = 0
|
||||
assert.equal((await retention.checkCooldownHorizon(2)).ok, true)
|
||||
})
|
||||
|
||||
// ── The sweep ──────────────────────────────────────────────────────────────
|
||||
|
||||
test('one tick sweeps all three tables at their own horizons', async () => {
|
||||
const now = new Date('2026-09-01T03:00:00Z')
|
||||
const result = await prune.tick(now)
|
||||
|
||||
assert.equal(daysBetween(now, world.calls.sends[0].before), 180)
|
||||
assert.equal(daysBetween(now, world.calls.cooldowns[0].before), 30)
|
||||
assert.equal(daysBetween(now, world.calls.outbox[0].before), 30)
|
||||
assert.deepEqual(result.warnings, [])
|
||||
})
|
||||
|
||||
test('each statement is bounded', async () => {
|
||||
await prune.tick(new Date())
|
||||
for (const table of ['cooldowns', 'outbox', 'sends']) {
|
||||
assert.equal(world.calls[table][0].limit, prune.BATCH, `${table} is batched`)
|
||||
}
|
||||
})
|
||||
|
||||
test('a sweep repeats while its batches come back full, and stops', async () => {
|
||||
world.deleted.sends = prune.BATCH
|
||||
const result = await prune.tick(new Date())
|
||||
assert.equal(world.calls.sends.length, prune.MAX_BATCHES, 'it stops rather than looping forever')
|
||||
assert.equal(result.sends, prune.BATCH * prune.MAX_BATCHES)
|
||||
})
|
||||
|
||||
test('one failing table does not stop the other two', async () => {
|
||||
patch(outboxDb, 'pruneTerminal', async () => { throw new Error('lock wait timeout') })
|
||||
const result = await prune.tick(new Date())
|
||||
assert.equal(result.outbox, 0)
|
||||
assert.equal(world.calls.cooldowns.length, 1, 'cooldowns still swept')
|
||||
assert.equal(world.calls.sends.length, 1, 'the send log still swept')
|
||||
})
|
||||
|
||||
test('the sweep runs even when the cooldown horizon is too short, and warns', async () => {
|
||||
// Deliberate: refusing to prune would trade a bounded, describable fault (one
|
||||
// rule may re-fire early) for the unbounded one this phase exists to end.
|
||||
world.longestCooldown = 86400
|
||||
world.settings.set('engagement_cooldowns_retain_days', '2')
|
||||
const result = await prune.tick(new Date())
|
||||
assert.equal(result.warnings.length, 0, 'two days clears a one-day cooldown')
|
||||
|
||||
world.longestCooldown = 30 * 86400 // longer than any rule can actually save
|
||||
const second = await prune.tick(new Date())
|
||||
assert.equal(second.warnings.length, 1)
|
||||
assert.equal(world.calls.cooldowns.length, 2, 'it swept anyway')
|
||||
})
|
||||
|
||||
test('an unreadable policy skips the run rather than sweeping on a guess', async () => {
|
||||
patch(retention, 'get', async () => { throw new Error('pool is dead') })
|
||||
const result = await prune.tick(new Date())
|
||||
assert.deepEqual(result, { cooldowns: 0, outbox: 0, sends: 0, warnings: [] })
|
||||
assert.equal(world.calls.sends.length, 0, 'nothing was deleted')
|
||||
})
|
||||
|
||||
test('start/stop is idempotent and leaves no live timer', () => {
|
||||
prune.start()
|
||||
prune.start()
|
||||
prune.stop()
|
||||
prune.stop()
|
||||
})
|
||||
|
||||
// The SQL these depend on — terminal-only, the give-up-before-reclaim order and
|
||||
// the batch LIMIT — is proved against a real server in engagementRetentionSql.test.js.
|
||||
// It cannot be proved here: the db modules destructure `query` at require time,
|
||||
// so there is nothing left to stub, and a hand-rolled stand-in would prove only
|
||||
// that two readings of the manual agree (which is exactly the trap Phase 4a's
|
||||
// `foundRows` defect sprang).
|
||||
318
server/test/engagementRetentionSql.test.js
Normal file
318
server/test/engagementRetentionSql.test.js
Normal file
@@ -0,0 +1,318 @@
|
||||
// ── The retention sweep, against a real MariaDB ────────────────────────────
|
||||
//
|
||||
// ENGAGEMENT.md Phase 14. `engagementRetention.test.js` proves the policy and
|
||||
// the worker's control flow against stubs, which is right for everything the
|
||||
// worker DECIDES. It cannot prove the three statements whose whole correctness
|
||||
// is what the server does with them, and this is the file for those — same shape
|
||||
// and same reasoning as `engagementEngineSql.test.js`, which exists because a
|
||||
// stub written from the same reading of the manual proves the reading, not the
|
||||
// server (that is how the `foundRows` cooldown defect got as far as it did).
|
||||
//
|
||||
// It is also Phase 14's acceptance rig, stated in the phase: *"a rig run shows
|
||||
// the sweep deleting terminal rows while leaving a `scheduled` outbox row and an
|
||||
// in-window send-log row alone."*
|
||||
//
|
||||
// **It SKIPS when there is no database**, deliberately: CI runs the suite with
|
||||
// the pool pointed at a dead port, and a file that failed there would make every
|
||||
// PR red for a reason unrelated to itself. Run it against this machine's
|
||||
// container with:
|
||||
//
|
||||
// DB_HOST=127.0.0.1 DB_PORT=3306 DB_USER=... DB_PASSWORD=... \
|
||||
// node --test test/engagementRetentionSql.test.js
|
||||
//
|
||||
// It creates its tables in a throwaway database named after the process and
|
||||
// drops it again, so it can never touch a real schema.
|
||||
|
||||
const { test, before, after } = require('node:test')
|
||||
const assert = require('node:assert/strict')
|
||||
const mariadb = require('mariadb')
|
||||
|
||||
// The three tables under sweep, with the two indexes Phase 14 added. Copied
|
||||
// rather than required, for the reason the engine's SQL file gives: requiring the
|
||||
// modules would drag in `utils/db`'s pool.
|
||||
const SCHEMA = `
|
||||
CREATE TABLE engagement_cooldowns (
|
||||
rule_id INT NOT NULL,
|
||||
user_id INT NOT NULL,
|
||||
subject_key VARCHAR(190) NOT NULL DEFAULT '',
|
||||
channel VARCHAR(32) NOT NULL DEFAULT '',
|
||||
last_fired_at DATETIME NOT NULL,
|
||||
fire_count INT NOT NULL DEFAULT 1,
|
||||
PRIMARY KEY (rule_id, user_id, subject_key, channel),
|
||||
INDEX idx_engc_sweep (last_fired_at)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
||||
CREATE TABLE engagement_outbox (
|
||||
id BIGINT AUTO_INCREMENT PRIMARY KEY,
|
||||
rule_id INT NOT NULL,
|
||||
trigger_id VARCHAR(96) NOT NULL,
|
||||
user_id INT NOT NULL,
|
||||
channel VARCHAR(32) NOT NULL,
|
||||
subject_key VARCHAR(190) NOT NULL DEFAULT '',
|
||||
payload JSON NOT NULL,
|
||||
dedupe_key VARCHAR(190) NULL,
|
||||
status ENUM('scheduled','sending','sent','failed','cancelled','suppressed') NOT NULL DEFAULT 'scheduled',
|
||||
due_at DATETIME NOT NULL,
|
||||
attempts SMALLINT NOT NULL DEFAULT 0,
|
||||
last_error TEXT NULL,
|
||||
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
|
||||
sent_at DATETIME NULL,
|
||||
INDEX idx_engo_due (status, due_at),
|
||||
INDEX idx_engo_sweep (status, created_at)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
||||
CREATE TABLE engagement_sends (
|
||||
id BIGINT AUTO_INCREMENT PRIMARY KEY,
|
||||
rule_id INT NULL,
|
||||
trigger_id VARCHAR(96) NOT NULL,
|
||||
user_id INT NULL,
|
||||
channel VARCHAR(32) NOT NULL,
|
||||
status ENUM('sent','failed','suppressed','bounced','complained') NOT NULL,
|
||||
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
INDEX idx_engs_rule_window (rule_id, created_at),
|
||||
INDEX idx_engs_sweep (created_at)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
||||
`
|
||||
|
||||
// Verbatim from the three `.db` files. If one of these drifts from its source the
|
||||
// test still passes and proves nothing, which is the standing cost of the copy —
|
||||
// the alternative (requiring the modules) costs a live pool on a dead port.
|
||||
const PRUNE_COOLDOWNS = 'DELETE FROM engagement_cooldowns WHERE last_fired_at < ? LIMIT ?'
|
||||
const PRUNE_OUTBOX = `
|
||||
DELETE FROM engagement_outbox
|
||||
WHERE status IN ('sent', 'failed', 'cancelled', 'suppressed')
|
||||
AND created_at < ?
|
||||
LIMIT ?`
|
||||
const PRUNE_SENDS = 'DELETE FROM engagement_sends WHERE created_at < ? LIMIT ?'
|
||||
const GIVE_UP = `
|
||||
UPDATE engagement_outbox
|
||||
SET status = 'failed', last_error = 'gave up after repeated interruptions'
|
||||
WHERE status = 'sending' AND updated_at < ? AND attempts >= ?`
|
||||
const RECLAIM = "UPDATE engagement_outbox SET status = 'scheduled' WHERE status = 'sending' AND updated_at < ?"
|
||||
|
||||
const DB = `rg_retain_test_${process.pid}`
|
||||
let pool = null
|
||||
let available = false
|
||||
|
||||
before(async () => {
|
||||
const admin = mariadb.createPool({
|
||||
host: process.env.DB_HOST || '127.0.0.1',
|
||||
port: Number(process.env.DB_PORT) || 3306,
|
||||
user: process.env.DB_USER || 'root',
|
||||
password: process.env.DB_PASSWORD || '',
|
||||
connectionLimit: 1,
|
||||
connectTimeout: 2000,
|
||||
initializationTimeout: 2000,
|
||||
multipleStatements: true,
|
||||
})
|
||||
try {
|
||||
await admin.query(`CREATE DATABASE ${DB}`)
|
||||
available = true
|
||||
} catch {
|
||||
available = false
|
||||
} finally {
|
||||
await admin.end().catch(() => {})
|
||||
}
|
||||
if (!available) return
|
||||
|
||||
pool = mariadb.createPool({
|
||||
host: process.env.DB_HOST || '127.0.0.1',
|
||||
port: Number(process.env.DB_PORT) || 3306,
|
||||
user: process.env.DB_USER || 'root',
|
||||
password: process.env.DB_PASSWORD || '',
|
||||
database: DB,
|
||||
connectionLimit: 3,
|
||||
multipleStatements: true,
|
||||
bigIntAsNumber: true,
|
||||
insertIdAsNumber: true,
|
||||
})
|
||||
await pool.query(SCHEMA)
|
||||
})
|
||||
|
||||
after(async () => {
|
||||
if (pool) {
|
||||
await pool.query(`DROP DATABASE IF EXISTS ${DB}`).catch(() => {})
|
||||
await pool.end().catch(() => {})
|
||||
}
|
||||
})
|
||||
|
||||
// Checked INSIDE each test, never as a `{ skip }` option — the option is
|
||||
// evaluated when the file is read, before `before()` has found out whether there
|
||||
// is a database, and every test skipped unconditionally looks like a pass.
|
||||
const SKIP = 'no database reachable - set DB_HOST/DB_PORT/DB_USER/DB_PASSWORD to run'
|
||||
const needDb = (t) => {
|
||||
if (available) return false
|
||||
t.skip(SKIP)
|
||||
return true
|
||||
}
|
||||
|
||||
const NOW = new Date('2026-09-01T12:00:00Z')
|
||||
const daysAgo = (d) => new Date(NOW.getTime() - d * 86400000)
|
||||
|
||||
const clear = async () => {
|
||||
await pool.query('DELETE FROM engagement_cooldowns')
|
||||
await pool.query('DELETE FROM engagement_outbox')
|
||||
await pool.query('DELETE FROM engagement_sends')
|
||||
}
|
||||
|
||||
const addOutbox = (status, createdAt, extra = {}) =>
|
||||
pool.query(
|
||||
`INSERT INTO engagement_outbox
|
||||
(rule_id, trigger_id, user_id, channel, payload, due_at, status, created_at, updated_at, attempts)
|
||||
VALUES (1, 't', 1, 'email', '{}', ?, ?, ?, ?, ?)`,
|
||||
[
|
||||
extra.dueAt || createdAt,
|
||||
status,
|
||||
createdAt,
|
||||
extra.updatedAt || createdAt,
|
||||
extra.attempts ?? 0,
|
||||
],
|
||||
)
|
||||
|
||||
// ── The acceptance rig ─────────────────────────────────────────────────────
|
||||
|
||||
test('the outbox sweep deletes terminal rows and leaves a scheduled one alone', async (t) => {
|
||||
if (needDb(t)) return
|
||||
await clear()
|
||||
|
||||
// Four terminal rows, well past the horizon.
|
||||
for (const status of ['sent', 'failed', 'cancelled', 'suppressed']) {
|
||||
await addOutbox(status, daysAgo(90))
|
||||
}
|
||||
// The row the phase names: old, and still a promise this deployment has made.
|
||||
await addOutbox('scheduled', daysAgo(90))
|
||||
// And one in flight, which is a worker's row and not the sweep's business.
|
||||
await addOutbox('sending', daysAgo(90))
|
||||
|
||||
const result = await pool.query(PRUNE_OUTBOX, [daysAgo(30), 1000])
|
||||
assert.equal(Number(result.affectedRows), 4, 'exactly the four terminal rows')
|
||||
|
||||
const left = await pool.query('SELECT status FROM engagement_outbox ORDER BY status')
|
||||
assert.deepEqual(left.map((r) => r.status), ['scheduled', 'sending'])
|
||||
})
|
||||
|
||||
test('an in-window row is left alone whatever its status', async (t) => {
|
||||
if (needDb(t)) return
|
||||
await clear()
|
||||
await addOutbox('sent', daysAgo(29))
|
||||
await addOutbox('sent', daysAgo(31))
|
||||
|
||||
const result = await pool.query(PRUNE_OUTBOX, [daysAgo(30), 1000])
|
||||
assert.equal(Number(result.affectedRows), 1)
|
||||
const [row] = await pool.query('SELECT created_at FROM engagement_outbox')
|
||||
assert.ok(row, 'the row inside the horizon survived')
|
||||
})
|
||||
|
||||
test('the send-log sweep leaves an in-window row alone', async (t) => {
|
||||
if (needDb(t)) return
|
||||
await clear()
|
||||
const add = (createdAt) => pool.query(
|
||||
`INSERT INTO engagement_sends (rule_id, trigger_id, user_id, channel, status, created_at)
|
||||
VALUES (1, 't', 1, 'email', 'sent', ?)`,
|
||||
[createdAt],
|
||||
)
|
||||
await add(daysAgo(200))
|
||||
await add(daysAgo(179))
|
||||
// The row the per-rule hourly ceiling counts. If a horizon could reach this,
|
||||
// the ceiling would silently stop capping anything.
|
||||
await add(new Date(NOW.getTime() - 60 * 1000))
|
||||
|
||||
const result = await pool.query(PRUNE_SENDS, [daysAgo(180), 1000])
|
||||
assert.equal(Number(result.affectedRows), 1)
|
||||
const [{ n }] = await pool.query('SELECT COUNT(*) AS n FROM engagement_sends')
|
||||
assert.equal(Number(n), 2)
|
||||
})
|
||||
|
||||
test('the cooldown sweep respects its batch limit and repeats', async (t) => {
|
||||
if (needDb(t)) return
|
||||
await clear()
|
||||
for (let i = 0; i < 5; i += 1) {
|
||||
await pool.query(
|
||||
`INSERT INTO engagement_cooldowns (rule_id, user_id, subject_key, channel, last_fired_at)
|
||||
VALUES (1, ?, '', 'email', ?)`,
|
||||
[i, daysAgo(90)],
|
||||
)
|
||||
}
|
||||
const first = await pool.query(PRUNE_COOLDOWNS, [daysAgo(30), 2])
|
||||
assert.equal(Number(first.affectedRows), 2, 'LIMIT bounds one statement')
|
||||
const second = await pool.query(PRUNE_COOLDOWNS, [daysAgo(30), 10])
|
||||
assert.equal(Number(second.affectedRows), 3, 'and the rest go on the next pass')
|
||||
})
|
||||
|
||||
// ── The bound the sweep depends on ─────────────────────────────────────────
|
||||
|
||||
test('a stale row that has burned its attempts is failed, not handed back', async (t) => {
|
||||
if (needDb(t)) return
|
||||
await clear()
|
||||
// Two rows stranded in 'sending' by a crash between the claim and the outcome.
|
||||
// One has attempts left; the other has spent them, and before Phase 14 nothing
|
||||
// could ever move it — MAX_ATTEMPTS is consulted only on a graceful `retry`,
|
||||
// so it cycled sending → scheduled → sending forever, never became terminal,
|
||||
// and was therefore never eligible for any sweep.
|
||||
await addOutbox('sending', daysAgo(1), { updatedAt: daysAgo(1), attempts: 1 })
|
||||
await addOutbox('sending', daysAgo(1), { updatedAt: daysAgo(1), attempts: 5 })
|
||||
|
||||
const gaveUp = await pool.query(GIVE_UP, [new Date(NOW.getTime() - 15 * 60 * 1000), 5])
|
||||
assert.equal(Number(gaveUp.affectedRows), 1, 'only the exhausted row')
|
||||
const reclaimed = await pool.query(RECLAIM, [new Date(NOW.getTime() - 15 * 60 * 1000)])
|
||||
assert.equal(Number(reclaimed.affectedRows), 1, 'and only the other one comes back')
|
||||
|
||||
const rows = await pool.query('SELECT status, attempts FROM engagement_outbox ORDER BY attempts')
|
||||
assert.deepEqual(rows.map((r) => r.status), ['scheduled', 'failed'])
|
||||
|
||||
// And now it is terminal, so the sweep can bound it.
|
||||
await pool.query("UPDATE engagement_outbox SET created_at = ? WHERE status = 'failed'", [daysAgo(90)])
|
||||
const swept = await pool.query(PRUNE_OUTBOX, [daysAgo(30), 1000])
|
||||
assert.equal(Number(swept.affectedRows), 1)
|
||||
})
|
||||
|
||||
test('reclaiming before giving up would loop forever — the order is the fix', async (t) => {
|
||||
if (needDb(t)) return
|
||||
await clear()
|
||||
await addOutbox('sending', daysAgo(1), { updatedAt: daysAgo(1), attempts: 5 })
|
||||
|
||||
// The wrong order, run deliberately: reclaim first, and the exhausted row is
|
||||
// back in 'scheduled' where findDue will pick it up again.
|
||||
await pool.query(RECLAIM, [new Date(NOW.getTime() - 15 * 60 * 1000)])
|
||||
const gaveUp = await pool.query(GIVE_UP, [new Date(NOW.getTime() - 15 * 60 * 1000), 5])
|
||||
assert.equal(Number(gaveUp.affectedRows), 0, 'nothing left in sending to fail')
|
||||
const [row] = await pool.query('SELECT status FROM engagement_outbox')
|
||||
assert.equal(row.status, 'scheduled', 'which is the forever-retry this ordering avoids')
|
||||
})
|
||||
|
||||
// ── The indexes ────────────────────────────────────────────────────────────
|
||||
|
||||
test('each sweep is an index range scan, not a table scan', async (t) => {
|
||||
if (needDb(t)) return
|
||||
await clear()
|
||||
// A plan on an empty table is not worth reading, so give the optimizer rows.
|
||||
for (let i = 0; i < 200; i += 1) {
|
||||
await addOutbox(i % 2 ? 'sent' : 'scheduled', daysAgo(i))
|
||||
await pool.query(
|
||||
`INSERT INTO engagement_sends (rule_id, trigger_id, user_id, channel, status, created_at)
|
||||
VALUES (1, 't', 1, 'email', 'sent', ?)`,
|
||||
[daysAgo(i)],
|
||||
)
|
||||
}
|
||||
await pool.query('ANALYZE TABLE engagement_outbox')
|
||||
await pool.query('ANALYZE TABLE engagement_sends')
|
||||
|
||||
const outboxPlan = await pool.query(
|
||||
`EXPLAIN SELECT id FROM engagement_outbox
|
||||
WHERE status IN ('sent','failed','cancelled','suppressed') AND created_at < ?`,
|
||||
[daysAgo(30)],
|
||||
)
|
||||
assert.ok(
|
||||
String(outboxPlan[0].key || '').includes('idx_engo'),
|
||||
`expected an index, got ${JSON.stringify(outboxPlan[0])}`,
|
||||
)
|
||||
|
||||
const sendsPlan = await pool.query(
|
||||
'EXPLAIN SELECT id FROM engagement_sends WHERE created_at < ?',
|
||||
[daysAgo(180)],
|
||||
)
|
||||
assert.equal(
|
||||
sendsPlan[0].key,
|
||||
'idx_engs_sweep',
|
||||
`every other index on this table has created_at in SECOND position: ${JSON.stringify(sendsPlan[0])}`,
|
||||
)
|
||||
})
|
||||
Reference in New Issue
Block a user