From e2d7dc4b82abac48a7c39859d8b063c30fd59fee Mon Sep 17 00:00:00 2001 From: jjbrunton Date: Tue, 11 Aug 2026 21:14:30 +0100 Subject: [PATCH] Prioritize growth retries and randomize intake --- docs/api.md | 1 + docs/configuration.md | 2 ++ docs/product-and-safety.md | 1 + docs/services.md | 4 ++-- src/database/growth-repository.ts | 30 +++++++++++++++++++++++++++++- src/services/account-growth.ts | 8 +++++++- src/web/api-catalog.ts | 3 +++ src/web/routes/growth.ts | 12 ++++++++++++ tests/account-growth.test.ts | 30 ++++++++++++++++++++++++++++++ tests/growth-repository.test.ts | 27 +++++++++++++++++++++++++++ tests/web/growth-route.test.ts | 28 ++++++++++++++++++++++++++++ 11 files changed, 142 insertions(+), 4 deletions(-) diff --git a/docs/api.md b/docs/api.md index 0ad6d23..af84407 100644 --- a/docs/api.md +++ b/docs/api.md @@ -110,6 +110,7 @@ an administrator credential. | `PATCH` | `/api/growth/{accountId}` | Enable, disable or re-pace comment growth for an account | admin | yes | | `POST` | `/api/growth/{accountId}/pause` | Stop an account from posting further comments | control | yes | | `POST` | `/api/growth/{accountId}/resume` | Clear a growth pause after reviewing the cause | admin | yes | +| `DELETE` | `/api/growth/{accountId}/queue` | Clear queued and retryable growth threads that have not been posted | admin | yes | | `POST` | `/api/growth/{accountId}/subreddits` | Add a community to an account's growth list | admin | yes | | `PATCH` | `/api/growth/{accountId}/subreddits/{subreddit}` | Change a growth community's weight, cap or enabled state | admin | yes | | `DELETE` | `/api/growth/{accountId}/subreddits/{subreddit}` | Remove a community from an account's growth list | admin | yes | diff --git a/docs/configuration.md b/docs/configuration.md index 2dabd82..c26bb8a 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -103,6 +103,8 @@ Operator-settable per account: enabled, the daily comment limit, minimum spacing A per-community `dailyCap` only distributes the account allowance; caps are not additive and cannot raise the account total. +Community weights feed a new weighted-random pick for each discovered growth thread. The grower queues only one new thread at a time, drains due retries and existing queue work before polling another community, and exposes an account-scoped queue clear that removes only unposted discovered/retryable rows. + ## Runtime-editable subset The dashboard/API exposes a deliberate subset: rate limits, dry-run, recency and ready-funnel limits, subreddit definitions/conversation behavior, and selected advanced configuration. Shared transport schemas live in `packages/api-contracts/src/config.ts`; the backend source schema is still the final validator. diff --git a/docs/product-and-safety.md b/docs/product-and-safety.md index a167fe3..3a4f0a4 100644 --- a/docs/product-and-safety.md +++ b/docs/product-and-safety.md @@ -65,6 +65,7 @@ This is inauthentic account activity under Reddit's content policy, and many sub - The daily allowance is adaptive by default: it starts at two comments per day and only rises while comments demonstrably survive and earn upvotes, and a removal or a heavily downvoted comment lowers it immediately. An operator may override the figure per account; doing so replaces the adaptive backoff but not the pause below, and the dashboard states that trade-off where the limit is set. - Two consecutive removals, a comment invisible to signed-out visitors, or a captcha challenge pauses the account until an operator clears it. A failed visibility probe proves nothing and never pauses anything. - One comment per thread across every owned account, a per-community daily cap, quiet hours, and jittered spacing. +- Due retries run before untouched queued threads, and any pending retry blocks new discovery until it is resolved or ages out. Discovery queues one thread from a fresh weighted-random community pick, avoiding predictable same-community batches. - Reddit's own refusals are authoritative: a community ban disables that community for that account, a throttle backs off at the scope Reddit named, and a signed-out response invalidates the session rather than being retried. - Growth deliberately continues while an account is held out of outbound messaging, because building the history is what lifts that hold. It is always the lowest-priority work an account performs, after inbound replies. - Dry run prepares and records a comment but never posts it. Every posted comment, and every comment refused by either gate, is kept with its reason in the growth audit trail. diff --git a/docs/services.md b/docs/services.md index ee04395..c5452bf 100644 --- a/docs/services.md +++ b/docs/services.md @@ -35,12 +35,12 @@ Comment-based growth exists so a new account accumulates the age/karma history R | Module | Owns | Important constraints | | --- | --- | --- | -| `services/account-growth.ts` | Per-account discovery, generation, posting, verification, and trust updates | Auto-posts without operator review, so both screening stages fail closed; an interrupted post is never retried. | +| `services/account-growth.ts` | Per-account discovery, generation, posting, verification, and trust updates | Auto-posts without operator review, so both screening stages fail closed; due retries and existing queue work run before discovery; an interrupted post is never retried. | | `services/growth-pacer.ts` | Adaptive allowance ladder, the operator limit override, promote/demote/pause decisions, quiet hours, spacing | Pure and clock-injected. Pause beats demote beats promote; an operator limit replaces the earned allowance but never the pause; jitter is derived from the previous post so the gate cannot flap. | | `services/growth-thread-screening.ts` | Pure thread eligibility (age window, comment count, locked/stickied/NSFW) | A comment on a dead or saturated thread wastes a scarce daily allowance. | | `services/growth-comment-safety.ts` | Deterministic pre-publication gate on generated comment text | Rejects links, contact solicitation, user tags, filler, and any text revealing the generator. Independent of prompt wording. | | `reddit/comment-client.ts` | Comment submission and comment-state lookups for one account | Classifies Reddit's error codes into retry/abandon/disable/re-auth/stop; runs inside the account's own browser session and shares the request coordinator. | -| `database/growth-repository.ts` | Growth settings, per-account community list, durable comment records | One comment per thread across every account; mutations require the expected worker lease. | +| `database/growth-repository.ts` | Growth settings, per-account community list, durable comment records | One comment per thread across every account; mutations require the expected worker lease; queue clearing deletes only unposted discovered/retryable rows. | Verification is the safety mechanism that matters most. Posted comments are swept at roughly 15 minutes, 2 hours, and 24 hours using one batched `info.json` request, and a single signed-out probe on the second sweep decides whether the comment is publicly visible at all. Two consecutive removals, or one invisible comment, hard-pauses the account until an operator clears it; a failed probe never pauses anything. diff --git a/src/database/growth-repository.ts b/src/database/growth-repository.ts index 403a134..a52c623 100644 --- a/src/database/growth-repository.ts +++ b/src/database/growth-repository.ts @@ -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, diff --git a/src/services/account-growth.ts b/src/services/account-growth.ts index 0941134..9714564 100644 --- a/src/services/account-growth.ts +++ b/src/services/account-growth.ts @@ -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; ` diff --git a/src/web/api-catalog.ts b/src/web/api-catalog.ts index 9e0926f..9b3fd7e 100644 --- a/src/web/api-catalog.ts +++ b/src/web/api-catalog.ts @@ -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, diff --git a/src/web/routes/growth.ts b/src/web/routes/growth.ts index e9f56d1..28cc340 100644 --- a/src/web/routes/growth.ts +++ b/src/web/routes/growth.ts @@ -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' }); diff --git a/tests/account-growth.test.ts b/tests/account-growth.test.ts index d0550c5..09a8d78 100644 --- a/tests/account-growth.test.ts +++ b/tests/account-growth.test.ts @@ -295,6 +295,36 @@ describe('AccountGrowthService', () => { expect(outcome).toMatchObject({ kind: 'idle' }); expect(h.postComment).not.toHaveBeenCalled(); }); + + it('waits for a queued retry instead of discovering another thread', async () => { + h.repo.enqueueDiscovered(ACCOUNT, [thread('retry')], NOW); + h.repo.claimNext(ACCOUNT, `growth:${ACCOUNT}`, 60_000, NOW); + h.repo.defer( + ACCOUNT, + 'retry', + `growth:${ACCOUNT}`, + 'Temporary generation failure', + NOW + 15 * 60_000, + NOW, + ); + + const outcome = await h.service.tick({ allowPosting: true }); + + expect(outcome).toEqual({ + kind: 'idle', + reason: 'Existing growth queue is waiting for its retry time', + }); + expect(h.getHotPosts).not.toHaveBeenCalled(); + expect(h.repo.listRecent(ACCOUNT)).toHaveLength(1); + }); + + it('queues only one thread so each new comment rerolls the subreddit', async () => { + h.getHotPosts.mockResolvedValue([thread('a'), thread('b'), thread('c')]); + + await h.service.tick({ allowPosting: true }); + + expect(h.repo.listRecent(ACCOUNT)).toHaveLength(1); + }); }); describe('posting', () => { diff --git a/tests/growth-repository.test.ts b/tests/growth-repository.test.ts index 1b01989..4c56fd5 100644 --- a/tests/growth-repository.test.ts +++ b/tests/growth-repository.test.ts @@ -295,6 +295,33 @@ describe('AccountGrowthRepository', () => { expect(repo.claimNext(ACCOUNT, 'worker-1', 60_000, 5_000)?.postId).toBe('a'); }); + it('claims a due retry before a newer untouched thread', () => { + repo.enqueueDiscovered(ACCOUNT, [ + thread('retry', { created_utc: 1_000 }), + thread('newer', { created_utc: 9_000 }), + ], 10_000); + repo.claimNext(ACCOUNT, 'worker-1', 60_000, 10_000); + repo.defer(ACCOUNT, 'newer', 'worker-1', 'Temporary failure', 11_000, 10_000); + + repo.enqueueDiscovered(ACCOUNT, [thread('newest', { created_utc: 12_000 })], 12_000); + + expect(repo.claimNext(ACCOUNT, 'worker-1', 60_000, 12_000)?.postId).toBe('newer'); + }); + + it('reports and clears only pending queue work', () => { + repo.enqueueDiscovered(ACCOUNT, [ + thread('queued', { created_utc: 1_000 }), + thread('posted', { created_utc: 2_000 }), + ], 1_000); + repo.claimNext(ACCOUNT, 'worker-1', 60_000, 1_000); + repo.defer(ACCOUNT, 'posted', 'worker-1', 'Temporary failure', 5_000, 1_500); + + expect(repo.hasPendingQueueWork(ACCOUNT)).toBe(true); + expect(repo.clearPendingQueue(ACCOUNT)).toBe(2); + expect(repo.hasPendingQueueWork(ACCOUNT)).toBe(false); + expect(repo.listRecent(ACCOUNT)).toEqual([]); + }); + it('rejects rather than retries a comment interrupted mid-post', () => { repo.enqueueDiscovered(ACCOUNT, [thread('a')], 1_000); repo.claimNext(ACCOUNT, 'worker-1', 60_000, 1_000); diff --git a/tests/web/growth-route.test.ts b/tests/web/growth-route.test.ts index a7f6a83..1fb6563 100644 --- a/tests/web/growth-route.test.ts +++ b/tests/web/growth-route.test.ts @@ -276,6 +276,34 @@ describe('growth route', () => { }); }); + describe('DELETE /api/growth/:accountId/queue', () => { + it('clears queued and retryable items without deleting posted history', async () => { + repo.enqueueDiscovered(ACCOUNT, [ + thread('queued'), + { ...thread('posted'), created_utc: 1_800_000 }, + ], 1_000); + repo.claimNext(ACCOUNT, 'worker', 60_000, 1_000); + repo.markPosted(ACCOUNT, 'posted', 'worker', { + generatedComment: 'A posted comment', + modelId: 'model', + commentFullname: 't1_posted', + commentPermalink: null, + nextVerifyAt: 5_000, + }, 2_000); + + const { status, json } = await call(app, 'DELETE', `/api/growth/${ENCODED}/queue`); + + expect(status).toBe(200); + expect(json).toMatchObject({ success: true, cleared: 1 }); + expect(repo.get(ACCOUNT, 'queued')).toBeNull(); + expect(repo.get(ACCOUNT, 'posted')?.status).toBe('posted'); + expect(pushed).toContainEqual({ + level: 'warn', + message: 'Cleared 1 pending growth item(s) for u/grower', + }); + }); + }); + describe('subreddit list', () => { it('adds a community and returns 201', async () => { const { status, json } = await call(app, 'POST', `/api/growth/${ENCODED}/subreddits`, {