diff --git a/apps/webapp/app/components/runs/RecentlyQueuedSection.tsx b/apps/webapp/app/components/runs/RecentlyQueuedSection.tsx new file mode 100644 index 000000000..ceeba61d5 --- /dev/null +++ b/apps/webapp/app/components/runs/RecentlyQueuedSection.tsx @@ -0,0 +1,51 @@ +import { DateTime } from "~/components/primitives/DateTime"; +import { Header3 } from "~/components/primitives/Headers"; +import { Paragraph } from "~/components/primitives/Paragraph"; + +export type RecentlyQueuedEntry = { + runId: string; + status: "QUEUED" | "DRAINING" | "FAILED" | "DONE"; + createdAt: string | Date; +}; + +// Runs the mollifier has buffered but the drainer hasn't yet materialised +// into Postgres. Without this surface they're invisible to the dashboard +// during the buffered window — the paginated runs list is PG-only. We +// render a compact header section so operators can see in-flight buffered +// entries at a glance while still scrolling the regular list below. +export function RecentlyQueuedSection({ entries }: { entries: RecentlyQueuedEntry[] }) { + if (entries.length === 0) return null; + + return ( +
+ Recently queued ({entries.length}) + + Triggers accepted into the burst buffer. They'll appear in the list below once the + drainer materialises them. + + +
+ ); +} diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs._index/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs._index/route.tsx index f555f9817..9909a798b 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs._index/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs._index/route.tsx @@ -41,6 +41,8 @@ import { useProject } from "~/hooks/useProject"; import { useSearchParams } from "~/hooks/useSearchParam"; import { useShortcutKeys } from "~/hooks/useShortcutKeys"; import { findProjectBySlug } from "~/models/project.server"; +import { getMollifierBuffer } from "~/v3/mollifier/mollifierBuffer.server"; +import { RecentlyQueuedSection } from "~/components/runs/RecentlyQueuedSection"; import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server"; import { getRunFiltersFromRequest } from "~/presenters/RunFilters.server"; import { NextRunListPresenter } from "~/presenters/v3/NextRunListPresenter.server"; @@ -94,6 +96,15 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => { ...filters, }); + // Mollifier buffer entries don't appear in the paginated PG query — they + // sit in Redis until the drainer materialises them. Surface them in a + // separate "Recently queued" section above the list so they're not + // invisible during the buffered window. + const mollifierBuffer = getMollifierBuffer(); + const recentlyQueued = mollifierBuffer + ? await mollifierBuffer.listEntriesForEnv(environment.id, 50).catch(() => []) + : []; + // Only persist rootOnly when no tasks are filtered. While a task filter is active, // the toggle's URL value can be a temporary auto-flip (or a user override scoped to // the current task filter), and we don't want either bleeding into the saved @@ -112,13 +123,18 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => { data: list, rootOnlyDefault: filters.rootOnly, filters, + recentlyQueued: recentlyQueued.map((entry) => ({ + runId: entry.runId, + status: entry.status, + createdAt: entry.createdAt, + })), }, headers ? { headers } : undefined ); }; export default function Page() { - const { data, rootOnlyDefault, filters } = useTypedLoaderData(); + const { data, rootOnlyDefault, filters, recentlyQueued } = useTypedLoaderData(); const { isConnected } = useDevPresence(); const project = useProject(); const environment = useEnvironment(); @@ -141,6 +157,7 @@ export default function Page() { + { }, ); }); + +describe("MollifierBuffer.listEntriesForEnv", () => { + redisTest( + "returns up to maxCount entries from the queue without consuming them", + { timeout: 20_000 }, + async ({ redisContainer }) => { + const buffer = new MollifierBuffer({ + redisOptions: { + host: redisContainer.getHost(), + port: redisContainer.getPort(), + password: redisContainer.getPassword(), + }, + entryTtlSeconds: 600, + logger: new Logger("test", "log"), + }); + + try { + await buffer.accept({ runId: "r1", envId: "env_a", orgId: "org_1", payload: "{}" }); + await buffer.accept({ runId: "r2", envId: "env_a", orgId: "org_1", payload: "{}" }); + await buffer.accept({ runId: "r3", envId: "env_a", orgId: "org_1", payload: "{}" }); + + const entries = await buffer.listEntriesForEnv("env_a", 2); + expect(entries).toHaveLength(2); + const runIds = entries.map((e) => e.runId); + expect(new Set(runIds).size).toBe(2); + for (const id of runIds) expect(["r1", "r2", "r3"]).toContain(id); + + // Non-destructive: the drainer can still pop all three. + const popped: string[] = []; + for (let i = 0; i < 3; i++) { + const entry = await buffer.pop("env_a"); + if (entry) popped.push(entry.runId); + } + expect(new Set(popped)).toEqual(new Set(["r1", "r2", "r3"])); + } finally { + await buffer.close(); + } + }, + ); + + redisTest("returns empty array when env queue is empty", { timeout: 20_000 }, async ({ redisContainer }) => { + const buffer = new MollifierBuffer({ + redisOptions: { + host: redisContainer.getHost(), + port: redisContainer.getPort(), + password: redisContainer.getPassword(), + }, + entryTtlSeconds: 600, + logger: new Logger("test", "log"), + }); + + try { + expect(await buffer.listEntriesForEnv("env_empty", 10)).toEqual([]); + } finally { + await buffer.close(); + } + }); + + redisTest("maxCount <= 0 returns empty without hitting redis", { timeout: 20_000 }, async ({ redisContainer }) => { + const buffer = new MollifierBuffer({ + redisOptions: { + host: redisContainer.getHost(), + port: redisContainer.getPort(), + password: redisContainer.getPassword(), + }, + entryTtlSeconds: 600, + logger: new Logger("test", "log"), + }); + + try { + expect(await buffer.listEntriesForEnv("env_a", 0)).toEqual([]); + expect(await buffer.listEntriesForEnv("env_a", -5)).toEqual([]); + } finally { + await buffer.close(); + } + }); +}); diff --git a/packages/redis-worker/src/mollifier/buffer.ts b/packages/redis-worker/src/mollifier/buffer.ts index f739e3ff3..6c0fbc453 100644 --- a/packages/redis-worker/src/mollifier/buffer.ts +++ b/packages/redis-worker/src/mollifier/buffer.ts @@ -128,6 +128,23 @@ export class MollifierBuffer { return this.redis.smembers(`mollifier:org-envs:${orgId}`); } + // Read-only listing of currently-queued entries for a single env. Used by + // the dashboard's "Recently queued" surface — LRANGE is non-destructive, + // so the drainer still pops these entries in order. Returns up to + // `maxCount` entries (the most-recently-queued ones, since accept LPUSHes + // onto the head). Each entry hash is fetched separately; a `null` from + // getEntry (TTL expired between LRANGE and HGETALL) is skipped. + async listEntriesForEnv(envId: string, maxCount: number): Promise { + if (maxCount <= 0) return []; + const runIds = await this.redis.lrange(`mollifier:queue:${envId}`, 0, maxCount - 1); + const entries: BufferEntry[] = []; + for (const runId of runIds) { + const entry = await this.getEntry(runId); + if (entry) entries.push(entry); + } + return entries; + } + async ack(runId: string): Promise { await this.redis.del(`mollifier:entries:${runId}`); }