mirror of
https://github.com/djdevin/recflare.git
synced 2026-10-27 16:08:58 -07:00
[rooms] clean up old votes
This commit is contained in:
@@ -6,9 +6,12 @@
|
|||||||
* record of what happened rather than a tally that gets rewritten — the tally is computed
|
* record of what happened rather than a tally that gets rewritten — the tally is computed
|
||||||
* from the rows.
|
* from the rows.
|
||||||
*
|
*
|
||||||
* The rows outlive the vote. Nothing deletes them when a session ends, and nothing needs to:
|
* The rows outlive the vote, but not by much: the `rooms` worker's cron deletes ballots
|
||||||
* a ballot is keyed to its `game_session_id`, and instance ids are not reused, so an old
|
* older than 5 minutes (its `ROOM_VOTE_RETENTION_MS`), which is the longest anything here
|
||||||
* session's votes can never be counted into a new one's.
|
* looks back — `VOTE_CALL_COOLDOWN_MS`. Nothing deletes them when a session ends, and
|
||||||
|
* nothing needs to: a ballot is keyed to its `game_session_id`, and instance ids are not
|
||||||
|
* reused, so an old session's votes can never be counted into a new one's. Raising the
|
||||||
|
* cooldown means raising that retention too.
|
||||||
*
|
*
|
||||||
* The `api` worker owns this schema/migration (migrations/0023_room_vote.sql and
|
* The `api` worker owns this schema/migration (migrations/0023_room_vote.sql and
|
||||||
* 0026_room_vote_init.sql, applied under
|
* 0026_room_vote_init.sql, applied under
|
||||||
|
|||||||
@@ -31,7 +31,7 @@ import { app as match, scheduled as matchScheduled } from '../../match/src/match
|
|||||||
import notify from '../../notify/src/notify.app'
|
import notify from '../../notify/src/notify.app'
|
||||||
import ns from '../../ns/src/ns.app'
|
import ns from '../../ns/src/ns.app'
|
||||||
import playersettings from '../../playersettings/src/playersettings.app'
|
import playersettings from '../../playersettings/src/playersettings.app'
|
||||||
import rooms from '../../rooms/src/rooms.app'
|
import { app as rooms, scheduled as roomsScheduled } from '../../rooms/src/rooms.app'
|
||||||
import storage from '../../storage/src/storage.app'
|
import storage from '../../storage/src/storage.app'
|
||||||
|
|
||||||
import type { Env } from './context'
|
import type { Env } from './context'
|
||||||
@@ -106,8 +106,15 @@ export default {
|
|||||||
return services[resolved.name].fetch(resolved.request, env, ctx)
|
return services[resolved.name].fetch(resolved.request, env, ctx)
|
||||||
},
|
},
|
||||||
|
|
||||||
// Only `match` runs a cron in the split deployment; this worker owns its presence sweep.
|
// `match` (presence sweep) and `rooms` (stale vote-to-kick ballots) each run a
|
||||||
scheduled(controller: ScheduledController, env: Env, ctx: ExecutionContext): Promise<void> | void {
|
// `*/5 * * * *` cron in the split deployment; this worker's one trigger fires both.
|
||||||
return matchScheduled(controller, env, ctx)
|
// Each schedules its work on `ctx.waitUntil`, so the runtime waits for both.
|
||||||
|
scheduled(
|
||||||
|
controller: ScheduledController,
|
||||||
|
env: Env,
|
||||||
|
ctx: ExecutionContext
|
||||||
|
): Promise<void> | void {
|
||||||
|
matchScheduled(controller, env, ctx)
|
||||||
|
roomsScheduled(controller, env, ctx)
|
||||||
},
|
},
|
||||||
} satisfies ExportedHandler<Env>
|
} satisfies ExportedHandler<Env>
|
||||||
|
|||||||
@@ -45,7 +45,8 @@
|
|||||||
"bindings": [{ "name": "RECFLARE_NOTIFICATIONS_HUB", "class_name": "NotificationsHub" }]
|
"bindings": [{ "name": "RECFLARE_NOTIFICATIONS_HUB", "class_name": "NotificationsHub" }]
|
||||||
},
|
},
|
||||||
"migrations": [{ "tag": "v1", "new_sqlite_classes": ["NotificationsHub"] }],
|
"migrations": [{ "tag": "v1", "new_sqlite_classes": ["NotificationsHub"] }],
|
||||||
// Presence sweep (owned by match). Every 5 minutes.
|
// Presence sweep (owned by match) and the stale vote-to-kick ballot sweep (owned by
|
||||||
|
// rooms) — both `*/5 * * * *`, so one trigger fires both. Every 5 minutes.
|
||||||
"triggers": {
|
"triggers": {
|
||||||
"crons": ["*/5 * * * *"]
|
"crons": ["*/5 * * * *"]
|
||||||
},
|
},
|
||||||
|
|||||||
@@ -37,6 +37,13 @@ Reads are the exception: they carry no envelope at all. A paged read (the save h
|
|||||||
`{ Results, TotalResults, TotalCount }` wrapper — `TotalResults` and `TotalCount` are the
|
`{ Results, TotalResults, TotalCount }` wrapper — `TotalResults` and `TotalCount` are the
|
||||||
same number, because the client's paged DTO and the reference disagree on the name.
|
same number, because the client's paged DTO and the reference disagree on the name.
|
||||||
|
|
||||||
|
## Cron
|
||||||
|
|
||||||
|
A `*/5 * * * *` cron (`scheduled`) deletes vote-to-kick ballots in `room_vote` older than
|
||||||
|
5 minutes. The table is the `api` worker's and append-only; a vote stays open for 60
|
||||||
|
seconds and the caller cooldown looks back 5 minutes, so nothing reads an older ballot.
|
||||||
|
The `mono` facade fires this sweep from its own trigger alongside `match`'s presence sweep.
|
||||||
|
|
||||||
## Development
|
## Development
|
||||||
|
|
||||||
### Run in dev mode
|
### Run in dev mode
|
||||||
|
|||||||
@@ -163,7 +163,7 @@ import {
|
|||||||
import type { Context } from 'hono'
|
import type { Context } from 'hono'
|
||||||
import type { RoomBan, RoomBanRecord, RoomPermission } from '@repo/domain'
|
import type { RoomBan, RoomBanRecord, RoomPermission } from '@repo/domain'
|
||||||
import type { MessageReceivedPayload } from '../../notify/src/notification-payloads'
|
import type { MessageReceivedPayload } from '../../notify/src/notification-payloads'
|
||||||
import type { App } from './context'
|
import type { App, Env } from './context'
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Room server. Rooms are stored in D1 as JSON blobs with generated columns for
|
* Room server. Rooms are stored in D1 as JSON blobs with generated columns for
|
||||||
@@ -4199,4 +4199,42 @@ app.get(
|
|||||||
)
|
)
|
||||||
)
|
)
|
||||||
|
|
||||||
export default app
|
/**
|
||||||
|
* How long a `room_vote` ballot is kept. The table is append-only (see the `api` worker's
|
||||||
|
* votes-db.ts, which owns the schema) and nothing there reads a ballot older than this:
|
||||||
|
* a vote stays open for 60 seconds (`VOTE_WINDOW_MS`) and the tally only counts ballots
|
||||||
|
* from the open vote's call on, and the caller throttle looks back exactly
|
||||||
|
* `VOTE_CALL_COOLDOWN_MS` — 5 minutes. Older rows are dead weight, so the cron drops them.
|
||||||
|
* Must not be shorter than that cooldown, or a purge would let a caller call again early.
|
||||||
|
*/
|
||||||
|
const ROOM_VOTE_RETENTION_MS = 5 * 60_000
|
||||||
|
|
||||||
|
/** Delete every ballot cast before `now - ROOM_VOTE_RETENTION_MS`. Returns rows removed. */
|
||||||
|
async function deleteStaleRoomVotes(db: D1Database, now = Date.now()): Promise<number> {
|
||||||
|
const res = await db
|
||||||
|
.prepare('DELETE FROM room_vote WHERE voted_at < ?1')
|
||||||
|
.bind(new Date(now - ROOM_VOTE_RETENTION_MS).toISOString())
|
||||||
|
.run()
|
||||||
|
return res.meta.changes ?? 0
|
||||||
|
}
|
||||||
|
|
||||||
|
/** Cron: sweep `room_vote` ballots older than {@link ROOM_VOTE_RETENTION_MS}. */
|
||||||
|
async function sweepStaleRoomVotes(env: Env): Promise<void> {
|
||||||
|
const removed = await deleteStaleRoomVotes(env.DB)
|
||||||
|
// The tagged logger is request-scoped (its middleware never runs for a cron), so
|
||||||
|
// log plainly here — Workers observability picks it up either way.
|
||||||
|
console.log(`room_vote sweep: removed ${removed} stale ballots`)
|
||||||
|
}
|
||||||
|
|
||||||
|
// The HTTP surface is a standard Hono app, exported by name so it can be mounted
|
||||||
|
// uniformly like every other worker (e.g. by the `mono` facade). The cron that sweeps
|
||||||
|
// stale vote-to-kick ballots is exported alongside it.
|
||||||
|
export { app }
|
||||||
|
|
||||||
|
export const scheduled: ExportedHandlerScheduledHandler<Env> = (_controller, env, ctx) => {
|
||||||
|
ctx.waitUntil(sweepStaleRoomVotes(env))
|
||||||
|
}
|
||||||
|
|
||||||
|
// Standalone entry: a Worker only runs `scheduled` when it's on the default export,
|
||||||
|
// so rooms keeps the object form the runtime requires to fire its `*/5 * * * *` cron.
|
||||||
|
export default { fetch: app.fetch, scheduled } satisfies ExportedHandler<Env>
|
||||||
|
|||||||
@@ -1,8 +1,13 @@
|
|||||||
import { adminSecretsStore, env, SELF } from 'cloudflare:test'
|
import {
|
||||||
|
adminSecretsStore,
|
||||||
|
createExecutionContext,
|
||||||
|
createScheduledController,
|
||||||
|
env,
|
||||||
|
SELF,
|
||||||
|
waitOnExecutionContext,
|
||||||
|
} from 'cloudflare:test'
|
||||||
import { beforeAll, describe, expect, it } from 'vitest'
|
import { beforeAll, describe, expect, it } from 'vitest'
|
||||||
|
|
||||||
import '../../rooms.app'
|
|
||||||
|
|
||||||
import {
|
import {
|
||||||
AUDIT_LOG_SCHEMA_DDL,
|
AUDIT_LOG_SCHEMA_DDL,
|
||||||
createRoomInstance,
|
createRoomInstance,
|
||||||
@@ -16,8 +21,10 @@ import {
|
|||||||
SUBROOM_SCHEMA_DDL,
|
SUBROOM_SCHEMA_DDL,
|
||||||
} from '@repo/domain'
|
} from '@repo/domain'
|
||||||
|
|
||||||
|
import { SCHEMA_DDL as ROOM_VOTE_SCHEMA_DDL } from '../../../../api/src/votes-db'
|
||||||
import { NotificationType } from '../../../../notify/src/notification-types'
|
import { NotificationType } from '../../../../notify/src/notification-types'
|
||||||
import importRooms from '../../../static/ImportRooms.json'
|
import importRooms from '../../../static/ImportRooms.json'
|
||||||
|
import { scheduled } from '../../rooms.app'
|
||||||
|
|
||||||
import type { Room } from '@repo/domain'
|
import type { Room } from '@repo/domain'
|
||||||
import type { Env } from '../../context'
|
import type { Env } from '../../context'
|
||||||
@@ -108,6 +115,8 @@ beforeAll(async () => {
|
|||||||
// The audit log the staff takedown writes to. Owned by `api`'s migrations; built here
|
// The audit log the staff takedown writes to. Owned by `api`'s migrations; built here
|
||||||
// directly, the way `notify`'s tests do.
|
// directly, the way `notify`'s tests do.
|
||||||
for (const stmt of AUDIT_LOG_SCHEMA_DDL) await env.DB.prepare(stmt).run()
|
for (const stmt of AUDIT_LOG_SCHEMA_DDL) await env.DB.prepare(stmt).run()
|
||||||
|
// Vote-to-kick ballots (owned by `api`) — the cron here sweeps the stale ones.
|
||||||
|
for (const stmt of ROOM_VOTE_SCHEMA_DDL) await env.DB.prepare(stmt).run()
|
||||||
// Seed each room and split its subrooms into the subroom table (mirrors 0007's backfill).
|
// Seed each room and split its subrooms into the subroom table (mirrors 0007's backfill).
|
||||||
for (const r of importRooms) await seedRoomWithSubRooms(env.DB, r as Record<string, unknown>)
|
for (const r of importRooms) await seedRoomWithSubRooms(env.DB, r as Record<string, unknown>)
|
||||||
|
|
||||||
@@ -5323,3 +5332,48 @@ describe('room name validation', () => {
|
|||||||
}
|
}
|
||||||
})
|
})
|
||||||
})
|
})
|
||||||
|
|
||||||
|
describe('room_vote sweep cron', () => {
|
||||||
|
async function castBallot(voterId: number, agoMs: number, init = 0): Promise<void> {
|
||||||
|
await env.DB.prepare(
|
||||||
|
`INSERT INTO room_vote (game_session_id, player_id, response, voter_id, voted_at, init)
|
||||||
|
VALUES (1, 205, 1, ?1, ?2, ?3)`
|
||||||
|
)
|
||||||
|
.bind(voterId, new Date(Date.now() - agoMs).toISOString(), init)
|
||||||
|
.run()
|
||||||
|
}
|
||||||
|
|
||||||
|
async function voterIds(): Promise<number[]> {
|
||||||
|
const rows = await env.DB.prepare('SELECT voter_id FROM room_vote ORDER BY voter_id').all<{
|
||||||
|
voter_id: number
|
||||||
|
}>()
|
||||||
|
return rows.results.map((r) => r.voter_id)
|
||||||
|
}
|
||||||
|
|
||||||
|
it('deletes ballots older than 5 minutes and keeps the rest', async () => {
|
||||||
|
await env.DB.prepare('DELETE FROM room_vote').run()
|
||||||
|
// A vote called long ago, and one that just closed: both dead weight.
|
||||||
|
await castBallot(1, 6 * 60_000, 1)
|
||||||
|
await castBallot(2, 5 * 60_000 + 1_000)
|
||||||
|
// A caller still inside their 5-minute cooldown, an open vote, and its answer.
|
||||||
|
await castBallot(3, 4 * 60_000 + 30_000, 1)
|
||||||
|
await castBallot(4, 30_000, 1)
|
||||||
|
await castBallot(5, 0)
|
||||||
|
|
||||||
|
// Driven through the module's own export rather than the `exports` proxy — a
|
||||||
|
// ScheduledController can't cross the isolate boundary the proxy serializes over.
|
||||||
|
const ctx = createExecutionContext()
|
||||||
|
await scheduled(createScheduledController(), env, ctx)
|
||||||
|
await waitOnExecutionContext(ctx)
|
||||||
|
|
||||||
|
expect(await voterIds()).toEqual([3, 4, 5])
|
||||||
|
})
|
||||||
|
|
||||||
|
it('is a no-op on an empty table', async () => {
|
||||||
|
await env.DB.prepare('DELETE FROM room_vote').run()
|
||||||
|
const ctx = createExecutionContext()
|
||||||
|
await scheduled(createScheduledController(), env, ctx)
|
||||||
|
await waitOnExecutionContext(ctx)
|
||||||
|
expect(await voterIds()).toEqual([])
|
||||||
|
})
|
||||||
|
})
|
||||||
|
|||||||
@@ -49,6 +49,12 @@
|
|||||||
"secret_name": "JWT_SECRET"
|
"secret_name": "JWT_SECRET"
|
||||||
}
|
}
|
||||||
],
|
],
|
||||||
|
// Vote-to-kick sweep. `room_vote` (owned by `api`) is append-only and a ballot
|
||||||
|
// matters for at most 5 minutes — a vote is open for 60s and the caller cooldown
|
||||||
|
// looks back 5 minutes — so the cron drops everything older. Every 5 minutes.
|
||||||
|
"triggers": {
|
||||||
|
"crons": ["*/5 * * * *"]
|
||||||
|
},
|
||||||
"upload_source_maps": true,
|
"upload_source_maps": true,
|
||||||
"observability": {
|
"observability": {
|
||||||
"enabled": false
|
"enabled": false
|
||||||
|
|||||||
Reference in New Issue
Block a user