Prioritize growth retries and randomize intake

This commit is contained in:
jjbrunton committed 2026-08-11 21:14:30 +01:00
1 parent b949b6ee37
commit e2d7dc4b82
11 files changed
+142 -4

No files matched your search

+29 -1
View File
@@ -460,7 +460,11 @@ export class AccountGrowthRepository {
const row = this.db.prepare(`
SELECT post_id FROM account_growth_comments
WHERE account_id = ? AND status IN ('discovered', 'retryable') AND available_at <= ?
ORDER BY thread_created_at DESC, discovered_at ASC
ORDER BY
CASE status WHEN 'retryable' THEN 0 ELSE 1 END,
CASE status WHEN 'retryable' THEN available_at END ASC,
thread_created_at DESC,
discovered_at ASC
LIMIT 1
`).get(accountId, now) as { post_id: string } | undefined;
if (!row) return null;
@@ -474,6 +478,30 @@ export class AccountGrowthRepository {
})();
}
/**
* Includes retries whose cooldown is still active. Discovery uses this to
* avoid growing the backlog while existing work is waiting to recover.
*/
hasPendingQueueWork(accountId: string): boolean {
return this.db.prepare(`
SELECT 1 FROM account_growth_comments
WHERE account_id = ? AND status IN ('discovered', 'retryable')
LIMIT 1
`).get(accountId) !== undefined;
}
/**
* Clears only work that has not reached Reddit. A posting lease is excluded
* because its remote outcome may be uncertain, and posted audit history is
* never touched.
*/
clearPendingQueue(accountId: string): number {
return this.db.prepare(`
DELETE FROM account_growth_comments
WHERE account_id = ? AND status IN ('discovered', 'retryable')
`).run(accountId).changes;
}
markPosted(
accountId: string,
postId: string,
+7 -1
View File
@@ -196,6 +196,9 @@ export class AccountGrowthService {
let claimed = this.repository.claimNext(this.accountId, this.workerId, LEASE_MS, now);
if (!claimed) {
if (this.repository.hasPendingQueueWork(this.accountId)) {
return { kind: 'idle', reason: 'Existing growth queue is waiting for its retry time' };
}
const discovery = await this.discover(signal);
if (discovery.kind === 'idle') return discovery;
claimed = this.repository.claimNext(this.accountId, this.workerId, LEASE_MS, this.now());
@@ -241,7 +244,10 @@ export class AccountGrowthService {
else rejections.set(screening.reason, (rejections.get(screening.reason) ?? 0) + 1);
}
const intake = this.repository.enqueueDiscovered(this.accountId, eligible, now, 10);
// Queue one thread at a time. Subreddit selection is randomized on every
// new unit of work, rather than one random community producing a long,
// predictable run of comments from a ten-thread batch.
const intake = this.repository.enqueueDiscovered(this.accountId, eligible, now, 1);
this.logger.info(
`Growth intake r/${subreddit.subreddit} for u/${this.options.redditUsername}: `
+ `${posts.length} fetched; ${eligible.length} eligible; ${intake.discovered} queued; `
+3
View File
@@ -189,6 +189,9 @@ export const API_OPERATIONS: readonly ApiOperation[] = [
access: 'admin',
responseSchema: GrowthAccountResponseSchema,
}),
operation('clearGrowthQueue', 'DELETE', '/api/growth/{accountId}/queue', 'Clear queued and retryable growth threads that have not been posted', ['Growth'], {
access: 'admin',
}),
operation('addGrowthSubreddit', 'POST', '/api/growth/{accountId}/subreddits', 'Add a community to an account\'s growth list', ['Growth'], {
access: 'admin',
requestSchema: GrowthSubredditCreateRequestSchema,
+12
View File
@@ -110,6 +110,18 @@ export function createGrowthRouter(ctx: WebContext): Router {
return res.json({ success: true, account: detail(repo, accountId, username) });
});
router.delete('/growth/:accountId/queue', (req, res) => {
const repo = growth();
if (!repo) return res.status(503).json({ error: 'Growth storage is unavailable' });
const { accountId } = req.params;
const username = accountName(accountId);
if (username === null) return res.status(404).json({ error: 'Reddit account not found' });
const cleared = repo.clearPendingQueue(accountId);
ctx.broadcaster.push('warn', `Cleared ${cleared} pending growth item(s) for u/${username}`);
return res.json({ success: true, cleared, account: detail(repo, accountId, username) });
});
router.post('/growth/:accountId/subreddits', (req, res) => {
const repo = growth();
if (!repo) return res.status(503).json({ error: 'Growth storage is unavailable' });