From 61b9dd9acf4c2917955d5e163e04d0cb18a34f35 Mon Sep 17 00:00:00 2001 From: vinzee Date: Mon, 12 Oct 2026 19:05:00 -0700 Subject: [PATCH 1/3] feat(api): add cursor pagination to external API v2 search Deep OFFSET paging on /api/v2/search makes ClickHouse produce and discard every skipped row, so page cost grows with page depth. Pages are now scoped to a time window instead of a growing offset. Send `cursor: "start"` to begin a walk, then pass back each response's `nextCursor` until it is null. The requested range is bucketed into progressively larger windows (15m, 6h, 6h, 12h, 24h) and a page queries exactly one of them, so page cost stops growing with depth. Windowing is opt-in. Omitting `cursor` keeps the previous single-range behaviour, because confining an offset-only request to the newest window would make a large `offset` skip within that window and never reach older rows. The cursor carries (windowIndex, offsetInWindow) and the resolved time range, plus a fingerprint of the query (source, filter, select, order). The fingerprint stops a cursor replayed against an edited query from returning plausible-looking pages from a different result set. The time range is carried rather than fingerprinted because `endTime` defaults to `now`: fingerprinting it rejected every cursor with a 400, and recomputing windows from it shifted every boundary mid-walk. Adjacent windows share a boundary instant and both date-range bounds default to inclusive, so a row landing on one would be returned on two consecutive pages. windowInclusivity() makes each window half-open while keeping the caller's own range start and end reachable. Windowing is skipped for toStartOf*/toDate() timestamp expressions, where the renderer forces both bounds inclusive and defeats that. A short page advances to the next window rather than querying a deeper offset in the current one, which addresses the API-side half of #1403. searchWindows moves from packages/app to common-utils so the API can import it; the app's behaviour is unchanged. A walk moves to the next window instead of emitting a cursor past MAX_CURSOR_OFFSET (1,000,000), windows only when the first sort key is the timestamp expression itself (a derived key such as toHour(Timestamp) would return windows out of order), and is limited to ranges of 366 days, checked on decoded cursors too since the pinned range travels in the unsigned cursor. Known limitation: ties at a page boundary within a window are still resolved arbitrarily, since pages use LIMIT/OFFSET inside the window. Offset paging's non-determinism on shared timestamps (#2614) is out of scope here. --- .changeset/search-cursor-pagination.md | 12 + packages/api/openapi.json | 14 +- .../external-api/__tests__/search.int.test.ts | 246 ++++++++++++++++++ .../api/src/routers/external-api/v2/search.ts | 118 ++++++++- .../v2/utils/__tests__/searchCursor.test.ts | 110 ++++++++ .../v2/utils/__tests__/searchWindows.test.ts | 192 ++++++++++++++ .../routers/external-api/v2/utils/search.ts | 111 +++++++- .../external-api/v2/utils/searchCursor.ts | 111 ++++++++ .../external-api/v2/utils/searchWindows.ts | 114 ++++++++ .../hooks/__tests__/useChartConfig.test.tsx | 6 +- packages/app/src/hooks/useChartConfig.tsx | 2 +- .../app/src/hooks/useOffsetPaginatedQuery.tsx | 14 +- .../src/__tests__/searchWindows.test.ts | 75 ++++++ .../src/core}/searchWindows.ts | 29 ++- 14 files changed, 1126 insertions(+), 28 deletions(-) create mode 100644 .changeset/search-cursor-pagination.md create mode 100644 packages/api/src/routers/external-api/v2/utils/__tests__/searchCursor.test.ts create mode 100644 packages/api/src/routers/external-api/v2/utils/__tests__/searchWindows.test.ts create mode 100644 packages/api/src/routers/external-api/v2/utils/searchCursor.ts create mode 100644 packages/api/src/routers/external-api/v2/utils/searchWindows.ts create mode 100644 packages/common-utils/src/__tests__/searchWindows.test.ts rename packages/{app/src/utils => common-utils/src/core}/searchWindows.ts (75%) diff --git a/.changeset/search-cursor-pagination.md b/.changeset/search-cursor-pagination.md new file mode 100644 index 0000000000..6c0b92073a --- /dev/null +++ b/.changeset/search-cursor-pagination.md @@ -0,0 +1,12 @@ +--- +'@hyperdx/api': minor +'@hyperdx/common-utils': patch +--- + +Add cursor pagination to the external API v2 search endpoint. Pass the +`nextCursor` from a response back as `cursor` to fetch the next page; pages are +scoped to a time window instead of a growing `OFFSET`, so page cost no longer +grows with depth. The existing `offset` parameter is unchanged. + +Keep requesting pages until `nextCursor` is `null` — a short page does not mean +the results are exhausted. diff --git a/packages/api/openapi.json b/packages/api/openapi.json index f6ea5912f6..ab147e52d3 100644 --- a/packages/api/openapi.json +++ b/packages/api/openapi.json @@ -4812,7 +4812,13 @@ "minimum": 0, "maximum": 10000, "default": 0, - "description": "Number of rows to skip (best-effort offset pagination). Default is\n0, max is 10000. Offset pagination is non-deterministic when\nmultiple rows share the same timestamp; for reliable deep paging\nfilter by the last Timestamp value returned in the previous page\ninstead of using a large offset.\n" + "description": "Number of rows to skip (best-effort offset pagination). Default is\n0, max is 10000. Offset pagination is non-deterministic when\nmultiple rows share the same timestamp. Prefer cursor for paging.\n" + }, + "cursor": { + "type": "string", + "maxLength": 1024, + "example": "start", + "description": "Send the literal string \"start\" to begin cursor paging, then pass\nback the nextCursor from each response. When set, offset is\nignored; omit cursor entirely to use offset-based paging.\nCursor paging supports time ranges of up to 366 days. As with\noffset, rows that share a timestamp can be skipped or repeated\nbetween pages unless orderBy breaks the tie.\n\nKeep requesting pages until nextCursor is null. A page that is\nempty or returns fewer rows than maxResults does NOT mean the\nresults are exhausted.\n" } } }, @@ -4834,6 +4840,12 @@ "rows": { "type": "integer", "description": "Number of rows in this response (not total matching rows)." + }, + "nextCursor": { + "type": "string", + "nullable": true, + "example": "eyJ2IjoxLCJ3IjoyLCJvIjo1MDAsImYiOiJhMWIyYzNkNGU1ZjZhN2I4In0", + "description": "Cursor for the next page, or null when the walk is complete or the request did not include a cursor. Pass it back as cursor in the next request to fetch the next page. This is an opaque value; no guarantees are made about its content.\n" } } }, diff --git a/packages/api/src/routers/external-api/__tests__/search.int.test.ts b/packages/api/src/routers/external-api/__tests__/search.int.test.ts index 29add43ec0..4864f5bec0 100644 --- a/packages/api/src/routers/external-api/__tests__/search.int.test.ts +++ b/packages/api/src/routers/external-api/__tests__/search.int.test.ts @@ -14,6 +14,10 @@ import Connection from '@/models/connection'; import type { ISource } from '@/models/source'; import { Source } from '@/models/source'; import type { IUser } from '@/models/user'; +import { + encodeSearchCursor, + fingerprintQuery, +} from '@/routers/external-api/v2/utils/searchCursor'; const DEFAULT_END_TIME = Math.floor(Date.now() / 60000) * 60000; const DEFAULT_START_TIME = DEFAULT_END_TIME - 3600 * 1000; @@ -557,4 +561,246 @@ describe('External API v2 Search', () => { expect(res.status).toBe(404); expect(res.body).toHaveProperty('message'); }); + + // Each test seeds its own rows: afterEach calls server.clearDBs(). + const seedCursorRows = async () => { + // 40 rows a minute apart, plus 3 on the newest 15m window's boundary. + const boundary = DEFAULT_END_TIME - 15 * 60 * 1000; + await bulkInsertLogs([ + ...Array.from({ length: 40 }, (_, i) => ({ + ServiceName: 'cursor-test', + Timestamp: new Date(DEFAULT_END_TIME - i * 60 * 1000), + SeverityText: 'INFO', + Body: `row-${i}`, + })), + ...Array.from({ length: 3 }, (_, i) => ({ + ServiceName: 'cursor-test', + Timestamp: new Date(boundary), + SeverityText: 'INFO', + Body: `boundary-${i}`, + })), + ]); + }; + + const cursorBody = () => ({ + sourceId: logSource.id.toString(), + startTime: iso(DEFAULT_START_TIME), + endTime: iso(DEFAULT_END_TIME), + where: 'ServiceName:"cursor-test"', + select: 'Timestamp,Body', + maxResults: 5, + }); + + it('walks every row exactly once with cursor pagination', async () => { + await seedCursorRows(); + + const seen: string[] = []; + let cursor: string | null | undefined; + let pages = 0; + for (; pages < 50; pages++) { + const res = await search({ + ...cursorBody(), + cursor: cursor ?? 'start', + }).expect(200); + seen.push(...res.body.data.map((r: Record) => r.Body)); + cursor = res.body.nextCursor; + if (cursor == null) break; + } + + expect(pages).toBeLessThan(49); + expect(cursor).toBeNull(); + expect(seen).toHaveLength(43); + expect(new Set(seen).size).toBe(43); + }); + + it.each([ + ['omitted', {}], + ['empty', { endTime: '' }], + ])( + 'pages with a defaulted endTime (%s), which resolves to a fresh now each call', + async (_label, endTimeField) => { + await seedCursorRows(); + + // An omitted or empty endTime resolves to a fresh `now` on each request. + const body = { + sourceId: logSource.id.toString(), + startTime: iso(DEFAULT_START_TIME), + ...endTimeField, + where: 'ServiceName:"cursor-test"', + select: 'Timestamp,Body', + maxResults: 5, + }; + + const seen: string[] = []; + let cursor: string | null | undefined; + let pages = 0; + for (; pages < 50; pages++) { + const res = await search({ ...body, cursor: cursor ?? 'start' }).expect( + 200, + ); + seen.push(...res.body.data.map((r: Record) => r.Body)); + cursor = res.body.nextCursor; + if (cursor == null) break; + } + + expect(cursor).toBeNull(); + expect(seen).toHaveLength(43); + expect(new Set(seen).size).toBe(43); + }, + ); + + it.each([ + ['on a whole second', 0], + ['mid-second', 500], + ])( + 'returns each row once around a window boundary %s with a second-precision timestamp column', + async (_label, boundaryMs) => { + const coarseSource = await Source.create({ + kind: SourceKind.Log, + team: logSource.team, + from: logSource.from, + defaultTableSelectExpression: 'Timestamp, SeverityText, Body', + timestampValueExpression: 'toDateTime(Timestamp), Timestamp', + connection: logSource.connection, + name: 'Logs (second-precision)', + }); + + // The first descending window boundary sits 15 minutes before endTime. + const endTime = DEFAULT_END_TIME + boundaryMs; + const boundary = endTime - 15 * 60 * 1000; + await bulkInsertLogs( + [-300, 0, 300].map(deltaMs => ({ + ServiceName: 'cursor-boundary-test', + Timestamp: new Date(boundary + deltaMs), + SeverityText: 'INFO', + Body: `at-${deltaMs}`, + })), + ); + + const body = { + sourceId: coarseSource.id.toString(), + startTime: iso(DEFAULT_START_TIME), + endTime: iso(endTime), + where: 'ServiceName:"cursor-boundary-test"', + select: 'Timestamp,Body', + orderBy: 'Timestamp DESC', + maxResults: 5, + }; + + const seen: string[] = []; + let cursor: string | null | undefined; + for (let pages = 0; pages < 50; pages++) { + const res = await search({ ...body, cursor: cursor ?? 'start' }).expect( + 200, + ); + seen.push(...res.body.data.map((r: Record) => r.Body)); + cursor = res.body.nextCursor; + if (cursor == null) break; + } + + expect(cursor).toBeNull(); + expect(seen.sort()).toEqual(['at--300', 'at-0', 'at-300']); + }, + ); + + it('rejects a cursor walk over a range longer than a year', async () => { + await search({ + ...cursorBody(), + startTime: iso(DEFAULT_END_TIME - 400 * 24 * 60 * 60 * 1000), + cursor: 'start', + }).expect(400); + }); + + const forgeCursor = ( + body: ReturnType & { orderBy: string }, + offset: number, + ) => + encodeSearchCursor( + { + windowIndex: 0, + offset, + startTime: iso(DEFAULT_START_TIME), + endTime: iso(DEFAULT_END_TIME), + }, + fingerprintQuery({ + sourceId: body.sourceId, + where: body.where, + whereLanguage: 'lucene', + select: body.select, + orderBy: body.orderBy, + }), + ); + + it('rejects a single-window cursor at the offset cap', async () => { + const body = { ...cursorBody(), orderBy: 'Body DESC' }; + const res = await search({ + ...body, + cursor: forgeCursor(body, 10_000), + }).expect(400); + expect(res.body.message).toMatch(/Too many matching rows/); + }); + + it('serves a short last page just under the offset cap', async () => { + const body = { ...cursorBody(), orderBy: 'Body DESC' }; + const res = await search({ + ...body, + cursor: forgeCursor(body, 9_998), + }).expect(200); + expect(res.body.nextCursor).toBeNull(); + }); + + it('rejects a cursor replayed with a different explicit range', async () => { + await seedCursorRows(); + + const first = await search({ ...cursorBody(), cursor: 'start' }).expect( + 200, + ); + expect(first.body.nextCursor).toBeTruthy(); + + await search({ + ...cursorBody(), + startTime: iso(DEFAULT_START_TIME + 60 * 1000), + cursor: first.body.nextCursor, + }).expect(400); + }); + + it('rejects a cursor reused against a different query', async () => { + await seedCursorRows(); + + const first = await search({ ...cursorBody(), cursor: 'start' }).expect( + 200, + ); + expect(first.body.nextCursor).toBeTruthy(); + + await search({ + ...cursorBody(), + where: 'ServiceName:"other-cursor-svc"', + cursor: first.body.nextCursor, + }).expect(400); + }); + + it('leaves offset pagination behaving as before', async () => { + await seedCursorRows(); + + // Rows span ~40 minutes, so a large offset must reach past the newest 15m window. + const first = await search({ ...cursorBody(), maxResults: 5 }).expect(200); + const deep = await search({ + ...cursorBody(), + maxResults: 5, + offset: 20, + }).expect(200); + + expect(first.body.data).toHaveLength(5); + expect(deep.body.data).toHaveLength(5); + // Offset-only callers are not driving a cursor walk. + expect(first.body.nextCursor).toBeNull(); + expect(deep.body.nextCursor).toBeNull(); + const firstBodies = first.body.data.map( + (r: Record) => r.Body, + ); + const deepBodies = deep.body.data.map( + (r: Record) => r.Body, + ); + expect(deepBodies).not.toEqual(firstBodies); + }); }); diff --git a/packages/api/src/routers/external-api/v2/search.ts b/packages/api/src/routers/external-api/v2/search.ts index ede7ed6676..417d2cb4f8 100644 --- a/packages/api/src/routers/external-api/v2/search.ts +++ b/packages/api/src/routers/external-api/v2/search.ts @@ -19,6 +19,15 @@ import logger from '@/utils/logger'; import { externalDashboardSearchRequestSchema } from '@/utils/zod'; import { runSearchConfig, type SearchErrorCode } from './utils/search'; +import { + CURSOR_START, + decodeSearchCursor, + encodeSearchCursor, + fingerprintQuery, + MAX_CURSOR_RANGE_MS, + MAX_SINGLE_WINDOW_OFFSET, + type SearchCursorState, +} from './utils/searchCursor'; const searchQueryDuration = getHistogram('hyperdx.search.query.duration_ms', { description: 'Duration of external API v2 search queries against ClickHouse.', @@ -132,9 +141,22 @@ const CH_USER_INPUT_ERRORS = new Set([ * description: | * Number of rows to skip (best-effort offset pagination). Default is * 0, max is 10000. Offset pagination is non-deterministic when - * multiple rows share the same timestamp; for reliable deep paging - * filter by the last Timestamp value returned in the previous page - * instead of using a large offset. + * multiple rows share the same timestamp. Prefer cursor for paging. + * cursor: + * type: string + * maxLength: 1024 + * example: "start" + * description: | + * Send the literal string "start" to begin cursor paging, then pass + * back the nextCursor from each response. When set, offset is + * ignored; omit cursor entirely to use offset-based paging. + * Cursor paging supports time ranges of up to 366 days. As with + * offset, rows that share a timestamp can be skipped or repeated + * between pages unless orderBy breaks the tie. + * + * Keep requesting pages until nextCursor is null. A page that is + * empty or returns fewer rows than maxResults does NOT mean the + * results are exhausted. * * SearchRow: * type: object @@ -155,6 +177,15 @@ const CH_USER_INPUT_ERRORS = new Set([ * rows: * type: integer * description: Number of rows in this response (not total matching rows). + * nextCursor: + * type: string + * nullable: true + * example: "eyJ2IjoxLCJ3IjoyLCJvIjo1MDAsImYiOiJhMWIyYzNkNGU1ZjZhN2I4In0" + * description: > + * Cursor for the next page, or null when the walk is complete or + * the request did not include a cursor. Pass it back as cursor in + * the next request to fetch the next page. This is an opaque + * value; no guarantees are made about its content. */ // Rejects semicolons and SELECT subqueries in column expressions. @@ -237,12 +268,21 @@ const searchRequestSchema = z.object({ .number() .int() .min(0) - .max(10_000) + .max(MAX_SINGLE_WINDOW_OFFSET) .optional() .default(0) .describe( 'Number of rows to skip for pagination (0-10000). Default: 0. ' + - 'Prefer timestamp-cursor pagination for large datasets.', + 'Prefer cursor for paging large datasets.', + ), + cursor: z + .string() + .max(1024) + .optional() + .describe( + 'Set to "start" to begin cursor paging, then pass back the nextCursor ' + + 'from each response. When set, offset is ignored. Keep requesting ' + + 'pages until nextCursor is null.', ), }); @@ -357,6 +397,9 @@ function codeToStatus(code: SearchErrorCode): number { case 'SOURCE_NOT_FOUND': case 'CONNECTION_NOT_FOUND': return 404; + case 'CURSOR_QUERY_MISMATCH': + case 'CURSOR_WINDOW_TOO_DENSE': + return 400; default: { const _exhaustive: never = code; void _exhaustive; @@ -383,6 +426,7 @@ router.post( orderBy, maxResults, offset, + cursor, } = req.body; const timeRange = parseTimeRange(startTime, endTime); @@ -391,6 +435,60 @@ router.post( } const { startDate, endDate } = timeRange; + const fingerprint = fingerprintQuery({ + sourceId, + where, + whereLanguage, + select, + orderBy: orderBy ?? '', + }); + + // The cursor pins this request's resolved range, so a defaulted endTime + // stays the same on later pages. + let cursorState: SearchCursorState | undefined; + if (cursor === CURSOR_START) { + cursorState = { + windowIndex: 0, + offset: 0, + startTime: startDate.toISOString(), + endTime: endDate.toISOString(), + }; + } else if (cursor != null) { + const decoded = decodeSearchCursor(cursor, fingerprint); + if ('error' in decoded) { + searchQueryErrors.add(1, { error_type: decoded.error }); + return res.status(400).json({ + message: + decoded.error === 'CURSOR_QUERY_MISMATCH' + ? 'Cursor does not match this query. Start a new search.' + : 'Cursor is malformed.', + }); + } + // Reject an explicitly changed bound; omitted or empty bounds default + // to `now` and are not compared. + const rangeChanged = + (!!startTime && startDate.toISOString() !== decoded.startTime) || + (!!endTime && endDate.toISOString() !== decoded.endTime); + if (rangeChanged) { + searchQueryErrors.add(1, { error_type: 'CURSOR_QUERY_MISMATCH' }); + return res.status(400).json({ + message: 'Cursor does not match this query. Start a new search.', + }); + } + cursorState = decoded; + } + if ( + cursorState && + Date.parse(cursorState.endTime) - Date.parse(cursorState.startTime) > + MAX_CURSOR_RANGE_MS + ) { + searchQueryErrors.add(1, { error_type: 'CURSOR_RANGE_TOO_LARGE' }); + return res.status(400).json({ + message: + 'Cursor paging supports time ranges up to 366 days. Narrow startTime/endTime.', + }); + } + const config = externalDashboardSearchRequestSchema.parse({ displayType: 'search' as const, sourceId, @@ -408,6 +506,7 @@ router.post( endDate, maxResults, offset, + cursorState, }), ); @@ -418,7 +517,14 @@ router.post( .json({ message: result.message }); } - return res.json({ data: result.data, rows: result.data.length }); + return res.json({ + data: result.data, + rows: result.data.length, + nextCursor: + result.nextCursorState == null + ? null + : encodeSearchCursor(result.nextCursorState, fingerprint), + }); } catch (err) { if (err instanceof ClickHouseQueryError) { const chType = ((err.cause as Record | undefined) diff --git a/packages/api/src/routers/external-api/v2/utils/__tests__/searchCursor.test.ts b/packages/api/src/routers/external-api/v2/utils/__tests__/searchCursor.test.ts new file mode 100644 index 0000000000..eb04787738 --- /dev/null +++ b/packages/api/src/routers/external-api/v2/utils/__tests__/searchCursor.test.ts @@ -0,0 +1,110 @@ +import { + decodeSearchCursor, + encodeSearchCursor, + fingerprintQuery, + MAX_CURSOR_OFFSET, +} from '@/routers/external-api/v2/utils/searchCursor'; + +const FP_INPUT = { + sourceId: '69b46cb0d964ce2d0b9506a8', + where: 'SeverityText:ERROR', + whereLanguage: 'lucene', + select: 'Timestamp,Body', + orderBy: 'Timestamp DESC', +}; + +const RANGE = { + startTime: '2026-05-10T00:00:00.000Z', + endTime: '2026-05-10T01:00:00.000Z', +}; + +describe('searchCursor', () => { + it('round-trips window index and offset', () => { + const fp = fingerprintQuery(FP_INPUT); + const cursor = encodeSearchCursor( + { windowIndex: 3, offset: 500, ...RANGE }, + fp, + ); + expect(decodeSearchCursor(cursor, fp)).toEqual({ + windowIndex: 3, + offset: 500, + ...RANGE, + }); + }); + + it('rejects a cursor minted for a different query', () => { + const fp = fingerprintQuery(FP_INPUT); + const cursor = encodeSearchCursor( + { windowIndex: 1, offset: 0, ...RANGE }, + fp, + ); + const otherFp = fingerprintQuery({ + ...FP_INPUT, + where: 'SeverityText:WARN', + }); + expect(decodeSearchCursor(cursor, otherFp)).toEqual({ + error: 'CURSOR_QUERY_MISMATCH', + }); + }); + + it('rejects malformed and unknown-version cursors', () => { + const fp = fingerprintQuery(FP_INPUT); + expect(decodeSearchCursor('not-base64!!', fp)).toEqual({ + error: 'INVALID_CURSOR', + }); + const future = Buffer.from( + JSON.stringify({ v: 99, w: 0, o: 0, f: fp }), + ).toString('base64url'); + expect(decodeSearchCursor(future, fp)).toEqual({ error: 'INVALID_CURSOR' }); + }); + + it('accepts in-window offsets past the request field cap', () => { + // A walk through a dense window passes the request field's 10,000 cap. + const fp = fingerprintQuery(FP_INPUT); + const deep = encodeSearchCursor( + { windowIndex: 0, offset: 50_000, ...RANGE }, + fp, + ); + expect(decodeSearchCursor(deep, fp)).toMatchObject({ offset: 50_000 }); + }); + + it('rejects an offset past the sanity bound', () => { + const fp = fingerprintQuery(FP_INPUT); + const over = encodeSearchCursor( + { windowIndex: 0, offset: MAX_CURSOR_OFFSET + 1, ...RANGE }, + fp, + ); + expect(decodeSearchCursor(over, fp)).toEqual({ error: 'INVALID_CURSOR' }); + + const atBound = encodeSearchCursor( + { windowIndex: 0, offset: MAX_CURSOR_OFFSET, ...RANGE }, + fp, + ); + expect(decodeSearchCursor(atBound, fp)).toMatchObject({ + offset: MAX_CURSOR_OFFSET, + }); + }); + + it('carries the time range instead of fingerprinting it', () => { + // A defaulted endTime differs on every request. + const fp = fingerprintQuery(FP_INPUT); + const cursor = encodeSearchCursor( + { windowIndex: 1, offset: 5, ...RANGE }, + fp, + ); + expect(decodeSearchCursor(cursor, fp)).toMatchObject(RANGE); + }); + + it('changes the fingerprint for every field that changes the result set', () => { + const base = fingerprintQuery(FP_INPUT); + for (const patch of [ + { sourceId: 'aaaaaaaaaaaaaaaaaaaaaaaa' }, + { where: 'x:1' }, + { whereLanguage: 'sql' }, + { select: 'Body' }, + { orderBy: 'Timestamp ASC' }, + ]) { + expect(fingerprintQuery({ ...FP_INPUT, ...patch })).not.toEqual(base); + } + }); +}); diff --git a/packages/api/src/routers/external-api/v2/utils/__tests__/searchWindows.test.ts b/packages/api/src/routers/external-api/v2/utils/__tests__/searchWindows.test.ts new file mode 100644 index 0000000000..9817283c0f --- /dev/null +++ b/packages/api/src/routers/external-api/v2/utils/__tests__/searchWindows.test.ts @@ -0,0 +1,192 @@ +import { resolveSearchWindows } from '@/routers/external-api/v2/utils/searchWindows'; + +const source = { timestampValueExpression: 'Timestamp' }; + +const START = new Date('2026-05-10T00:00:00.000Z'); +const END = new Date('2026-05-11T00:00:00.000Z'); + +describe('resolveSearchWindows', () => { + it('windows a timestamp-led descending order', async () => { + const windows = await resolveSearchWindows( + source, + 'Timestamp DESC', + START, + END, + ); + expect(windows.length).toBeGreaterThan(1); + expect(windows[0].direction).toBe('DESC'); + expect(windows[0].endTime).toEqual(END); + }); + + it('windows a timestamp-led ascending order in the other direction', async () => { + const windows = await resolveSearchWindows( + source, + 'Timestamp ASC', + START, + END, + ); + expect(windows[0].direction).toBe('ASC'); + expect(windows[0].startTime).toEqual(START); + }); + + it('disengages windowing when the order does not lead with the timestamp', async () => { + const windows = await resolveSearchWindows( + source, + 'SeverityText DESC', + START, + END, + ); + expect(windows).toHaveLength(1); + expect(windows[0].startTime).toEqual(START); + expect(windows[0].endTime).toEqual(END); + }); + + it('returns exactly one window when start equals end', async () => { + const windows = await resolveSearchWindows( + source, + 'Timestamp DESC', + START, + START, + ); + expect(windows).toHaveLength(1); + }); + + it('covers the full range with no gaps between adjacent windows', async () => { + const windows = await resolveSearchWindows( + source, + 'Timestamp DESC', + START, + END, + ); + windows + .slice(0, -1) + .forEach((w, i) => + expect(w.startTime).toEqual(windows.at(i + 1)?.endTime), + ); + expect(windows[windows.length - 1].startTime).toEqual(START); + expect(windows[0].endTime).toEqual(END); + }); + + it.each([ + ['descending', 'Timestamp DESC'], + ['ascending', 'Timestamp ASC'], + ])( + 'snaps %s interior boundaries to whole seconds and keeps the caller bounds exact', + async (_, orderBy) => { + const start = new Date('2026-05-10T00:00:00.250Z'); + const end = new Date('2026-05-11T00:00:00.750Z'); + const windows = await resolveSearchWindows(source, orderBy, start, end); + expect(windows.length).toBeGreaterThan(1); + + const bounds = windows.flatMap(w => [w.startTime, w.endTime]); + const interior = bounds.filter( + t => t.getTime() !== start.getTime() && t.getTime() !== end.getTime(), + ); + expect(interior.length).toBeGreaterThan(0); + expect(interior.every(t => t.getTime() % 1000 === 0)).toBe(true); + expect(bounds.map(t => t.getTime())).toContain(start.getTime()); + expect(bounds.map(t => t.getTime())).toContain(end.getTime()); + windows.forEach((w, i) => expect(w.windowIndex).toBe(i)); + }, + ); + + it('drops a window that snapping leaves empty and keeps the range covered', async () => { + const start = new Date('2026-05-10T00:00:00.400Z'); + const end = new Date('2026-05-10T00:15:00.700Z'); + const windows = await resolveSearchWindows( + source, + 'Timestamp DESC', + start, + end, + ); + + expect(windows).toEqual([ + { startTime: start, endTime: end, windowIndex: 0, direction: 'DESC' }, + ]); + }); + + it.each([ + [ + 'a compound toStartOf part next to the raw column', + 'toStartOfMinute(Timestamp), Timestamp', + undefined, + ], + [ + 'a Date column next to a DateTime64 column', + 'EventDate, Timestamp', + 'Date', + ], + ])('windows %s', async (_, timestampValueExpression, eventDateType) => { + const windows = await resolveSearchWindows( + { timestampValueExpression }, + 'Timestamp DESC', + START, + END, + { + columnType: async column => + column === 'EventDate' ? eventDateType : 'DateTime64(9)', + }, + ); + expect(windows.length).toBeGreaterThan(1); + }); + + it.each([ + [ + 'a lone toStartOf expression', + 'toStartOfMinute(Timestamp)', + 'toStartOfMinute(Timestamp) DESC', + undefined, + ], + ['a bare Date column', 'EventDate', 'EventDate DESC', 'Date'], + [ + 'a Nullable(Date32) column', + 'EventDate', + 'EventDate DESC', + 'Nullable(Date32)', + ], + ])( + 'uses one window for %s, whose bounds are inclusive on both ends', + async (_, timestampValueExpression, orderBy, columnType) => { + const windows = await resolveSearchWindows( + { timestampValueExpression }, + orderBy, + START, + END, + { columnType: async () => columnType }, + ); + expect(windows).toHaveLength(1); + }, + ); + + it('windows a default tuple order led by the timestamp', async () => { + const tupleSource = { timestampValueExpression: 'TimestampTime' }; + const windows = await resolveSearchWindows( + tupleSource, + '(TimestampTime, Timestamp) DESC', + START, + END, + ); + expect(windows.length).toBeGreaterThan(1); + expect(windows[0].direction).toBe('DESC'); + }); + + it('uses one window when a tuple order is not led by the timestamp', async () => { + const windows = await resolveSearchWindows( + source, + '(SeverityText, Timestamp) DESC', + START, + END, + ); + expect(windows).toHaveLength(1); + }); + + it('uses one window when the first sort key only derives from the timestamp', async () => { + const windows = await resolveSearchWindows( + source, + 'toHour(Timestamp) DESC', + START, + END, + ); + expect(windows).toHaveLength(1); + }); +}); diff --git a/packages/api/src/routers/external-api/v2/utils/search.ts b/packages/api/src/routers/external-api/v2/utils/search.ts index 552fdb14be..9d888bfb21 100644 --- a/packages/api/src/routers/external-api/v2/utils/search.ts +++ b/packages/api/src/routers/external-api/v2/utils/search.ts @@ -1,5 +1,6 @@ import { getMetadata } from '@hyperdx/common-utils/dist/core/metadata'; import { buildSearchChartConfig } from '@hyperdx/common-utils/dist/core/searchChartConfig'; +import { windowInclusivity } from '@hyperdx/common-utils/dist/core/searchWindows'; import { getFirstTimestampValueExpression, splitAndTrimWithBracket, @@ -19,7 +20,18 @@ import { getConnectionById } from '@/controllers/connection'; import { getSource } from '@/controllers/sources'; import type { ExternalDashboardSearchRequestConfig } from '@/utils/zod'; -export type SearchErrorCode = 'SOURCE_NOT_FOUND' | 'CONNECTION_NOT_FOUND'; +import { + MAX_CURSOR_OFFSET, + MAX_SINGLE_WINDOW_OFFSET, + type SearchCursorState, +} from './searchCursor'; +import { resolveSearchWindows } from './searchWindows'; + +export type SearchErrorCode = + | 'SOURCE_NOT_FOUND' + | 'CONNECTION_NOT_FOUND' + | 'CURSOR_QUERY_MISMATCH' + | 'CURSOR_WINDOW_TOO_DENSE'; type SearchError = { isError: true; @@ -30,6 +42,7 @@ type SearchError = { type SearchResults = { isError: false; data: Record[]; + nextCursorState: SearchCursorState | null; }; // Mirrors `optimizeDefaultOrderBy` + `useDefaultOrderBy` from DBSearchPage.tsx. @@ -85,6 +98,7 @@ export async function runSearchConfig({ endDate, maxResults, offset, + cursorState, }: { teamId: string; config: ExternalDashboardSearchRequestConfig; @@ -92,6 +106,7 @@ export async function runSearchConfig({ endDate: Date; maxResults: number; offset: number; + cursorState?: SearchCursorState; }): Promise { const source = await getSource(teamId, config.sourceId); if (!source) { @@ -133,22 +148,81 @@ export async function runSearchConfig({ ...(requestTimeout != null ? { requestTimeout } : {}), }); + const effectiveOrderBy = + config.orderBy?.trim() || resolveSearchOrderBy(source); + + // Windows come from the range pinned in the cursor, so boundaries stay put + // between pages. + const walkStart = cursorState ? new Date(cursorState.startTime) : startDate; + const walkEnd = cursorState ? new Date(cursorState.endTime) : endDate; + + const metadata = getMetadata(clickhouseClient); + + // Offset-only requests use one window over the whole range. + const windows = await resolveSearchWindows( + source, + effectiveOrderBy, + walkStart, + walkEnd, + { + forceSingleWindow: !cursorState, + columnType: async column => + ( + await metadata.getColumn({ + databaseName: source.from.databaseName, + tableName: source.from.tableName, + column, + connectionId: source.connection.toString(), + }) + )?.type, + }, + ); + const windowIndex = cursorState?.windowIndex ?? 0; + const window = windows[windowIndex]; + if (window == null) { + return { + isError: true, + code: 'CURSOR_QUERY_MISMATCH', + message: 'Cursor refers to a page outside the current time range', + }; + } + const effectiveOffset = cursorState ? cursorState.offset : offset; + // A single window keeps the `offset` field's cap. Erroring, rather than + // ending the walk, keeps a truncated result from looking complete. + const offsetCap = + windows.length === 1 ? MAX_SINGLE_WINDOW_OFFSET : MAX_CURSOR_OFFSET; + const pageLimit = cursorState + ? Math.min(maxResults, offsetCap - effectiveOffset) + : maxResults; + if (cursorState && pageLimit <= 0) { + return { + isError: true, + code: 'CURSOR_WINDOW_TOO_DENSE', + message: + 'Too many matching rows to page further. Narrow the time range or the filter.', + }; + } + const searchBase = buildSearchChartConfig(source, { where: typeof config.where === 'string' ? config.where : '', whereLanguage: config.whereLanguage ?? 'lucene', select: config.select ?? null, displayType: DisplayType.Search, - orderBy: config.orderBy?.trim() || resolveSearchOrderBy(source), - dateRange: [startDate, endDate], + orderBy: effectiveOrderBy, + dateRange: [window.startTime, window.endTime], + ...windowInclusivity(window, windows.length), }); const chartConfig: ChartConfigWithDateRange = { ...searchBase, connection: source.connection.toString(), - limit: { limit: maxResults, offset }, + // A cursor walk reads one extra row to learn whether the window continues. + limit: { + limit: cursorState ? pageLimit + 1 : maxResults, + offset: effectiveOffset, + }, } as ChartConfigWithDateRange; - const metadata = getMetadata(clickhouseClient); const result = await clickhouseClient.queryChartConfig({ config: chartConfig, metadata, @@ -164,8 +238,27 @@ export async function runSearchConfig({ throw new Error('Unexpected ClickHouse response shape: missing data array'); } - return { - isError: false, - data: (result as { data: unknown[] }).data as Record[], - }; + const rows = (result as { data: unknown[] }).data as Record< + string, + unknown + >[]; + const hasMoreInWindow = cursorState != null && rows.length > pageLimit; + const data = cursorState ? rows.slice(0, pageLimit) : rows; + + // Only a cursor walk gets a next cursor; offset-only callers are unchanged. + let nextCursorState: SearchCursorState | null = null; + if (cursorState) { + const pinned = { + startTime: cursorState.startTime, + endTime: cursorState.endTime, + }; + const nextOffset = effectiveOffset + data.length; + if (hasMoreInWindow) { + nextCursorState = { windowIndex, offset: nextOffset, ...pinned }; + } else if (windowIndex + 1 < windows.length) { + nextCursorState = { windowIndex: windowIndex + 1, offset: 0, ...pinned }; + } + } + + return { isError: false, data, nextCursorState }; } diff --git a/packages/api/src/routers/external-api/v2/utils/searchCursor.ts b/packages/api/src/routers/external-api/v2/utils/searchCursor.ts new file mode 100644 index 0000000000..eebcf1ef1d --- /dev/null +++ b/packages/api/src/routers/external-api/v2/utils/searchCursor.ts @@ -0,0 +1,111 @@ +import { createHash } from 'crypto'; + +const CURSOR_VERSION = 1; +const FINGERPRINT_LENGTH = 16; +// Above the `offset` field's 10,000 cap so a walk can page through a dense +// window; still bounded because cursors are unsigned. +export const MAX_CURSOR_OFFSET = 1_000_000; +// Also the `offset` request field's cap. +export const MAX_SINGLE_WINDOW_OFFSET = 10_000; + +// Windows are built up front, so the range is capped. Also checked on decoded +// cursors, which are unsigned. +export const MAX_CURSOR_RANGE_MS = 366 * 24 * 60 * 60 * 1000; + +/** Literal `cursor` value that starts a windowed cursor walk. */ +export const CURSOR_START = 'start'; + +export type SearchCursorState = { + windowIndex: number; + offset: number; + /** Time range pinned by the walk, so a defaulted endTime stays stable. */ + startTime: string; + endTime: string; +}; + +export type CursorFingerprintInput = { + sourceId: string; + where: string; + whereLanguage: string; + select: string; + orderBy: string; +}; + +export type CursorDecodeError = { + error: 'INVALID_CURSOR' | 'CURSOR_QUERY_MISMATCH'; +}; + +/** + * Ties a cursor to the query that produced it, so it can't be reused after + * `where`, `select`, or `orderBy` change. The time range travels in the cursor + * instead, so a defaulted `endTime` doesn't invalidate it. + */ +export function fingerprintQuery(input: CursorFingerprintInput): string { + // A JSON array keeps field boundaries unambiguous. + const canonical = JSON.stringify([ + input.sourceId, + input.where, + input.whereLanguage, + input.select, + input.orderBy, + ]); + return createHash('sha256') + .update(canonical) + .digest('hex') + .slice(0, FINGERPRINT_LENGTH); +} + +export function encodeSearchCursor( + state: SearchCursorState, + fingerprint: string, +): string { + return Buffer.from( + JSON.stringify({ + v: CURSOR_VERSION, + w: state.windowIndex, + o: state.offset, + f: fingerprint, + s: state.startTime, + e: state.endTime, + }), + ).toString('base64url'); +} + +export function decodeSearchCursor( + cursor: string, + fingerprint: string, +): SearchCursorState | CursorDecodeError { + let parsed: unknown; + try { + parsed = JSON.parse(Buffer.from(cursor, 'base64url').toString('utf8')); + } catch { + return { error: 'INVALID_CURSOR' }; + } + + if (parsed == null || typeof parsed !== 'object') { + return { error: 'INVALID_CURSOR' }; + } + const fields: Record = { ...parsed }; + const { v, w, o, f, s: start, e: end } = fields; + if ( + v !== CURSOR_VERSION || + typeof w !== 'number' || + typeof o !== 'number' || + !Number.isInteger(w) || + !Number.isInteger(o) || + w < 0 || + o < 0 || + o > MAX_CURSOR_OFFSET || + typeof f !== 'string' || + typeof start !== 'string' || + typeof end !== 'string' || + Number.isNaN(Date.parse(start)) || + Number.isNaN(Date.parse(end)) + ) { + return { error: 'INVALID_CURSOR' }; + } + if (f !== fingerprint) { + return { error: 'CURSOR_QUERY_MISMATCH' }; + } + return { windowIndex: w, offset: o, startTime: start, endTime: end }; +} diff --git a/packages/api/src/routers/external-api/v2/utils/searchWindows.ts b/packages/api/src/routers/external-api/v2/utils/searchWindows.ts new file mode 100644 index 0000000000..fc85fddf29 --- /dev/null +++ b/packages/api/src/routers/external-api/v2/utils/searchWindows.ts @@ -0,0 +1,114 @@ +import { + generateTimeWindowsAscending, + generateTimeWindowsDescending, + TimeWindow, +} from '@hyperdx/common-utils/dist/core/searchWindows'; +import { + classifyTimestampType, + getFirstOrderingItem, + isFirstOrderByAscending, + parseToStartOfFunction, + removeTrailingDirection, + splitAndTrimWithBracket, +} from '@hyperdx/common-utils/dist/core/utils'; + +// Windows keep the requested order only when the first sort key is the +// timestamp itself, not a derived key such as toHour(Timestamp). Stricter than +// isTimestampExpressionInFirstOrderBy, which accepts any key containing it. +function ordersByTimestamp( + orderBy: string, + timestampValueExpression: string | undefined, +): boolean { + const first = getFirstOrderingItem(orderBy); + if (typeof first !== 'string' || timestampValueExpression == null) { + return false; + } + const expression = removeTrailingDirection(first); + // A tuple order such as (TimestampTime, Timestamp) is led by its first element. + const leading = /^\(.*\)$/s.test(expression) + ? splitAndTrimWithBracket(expression.slice(1, -1))[0] + : expression; + return splitAndTrimWithBracket(timestampValueExpression).includes(leading); +} + +// toStartOf*, toDate and Date-typed parts get inclusive bounds on both ends. +// Windows stay half-open only if another part compares exact times. +async function keepsExactTime( + timestampValueExpression: string, + columnType: (column: string) => Promise, +): Promise { + for (const part of splitAndTrimWithBracket(timestampValueExpression)) { + if (parseToStartOfFunction(part) || /^toDate\s*\(/i.test(part)) continue; + if (classifyTimestampType(await columnType(part))?.kind === 'date') + continue; + return true; + } + return false; +} + +export async function resolveSearchWindows( + source: { timestampValueExpression?: string }, + orderBy: string, + startDate: Date, + endDate: Date, + { + forceSingleWindow = false, + columnType = async () => undefined, + }: { + forceSingleWindow?: boolean; + columnType?: (column: string) => Promise; + } = {}, +): Promise { + const canWindow = + !forceSingleWindow && + ordersByTimestamp(orderBy, source.timestampValueExpression) && + (await keepsExactTime(source.timestampValueExpression ?? '', columnType)); + + if (!canWindow) { + return [ + { + startTime: startDate, + endTime: endDate, + windowIndex: 0, + direction: isFirstOrderByAscending(orderBy) ? 'ASC' : 'DESC', + }, + ]; + } + + const windows = isFirstOrderByAscending(orderBy) + ? generateTimeWindowsAscending(startDate, endDate) + : generateTimeWindowsDescending(startDate, endDate); + return snapBoundariesToSeconds(windows, startDate, endDate); +} + +// Floors boundaries between windows to the second, so a second-precision +// column such as TimestampTime doesn't drop rows at a boundary. The caller's +// bounds stay exact; windows left empty are dropped. +function snapBoundariesToSeconds( + windows: TimeWindow[], + startDate: Date, + endDate: Date, +): TimeWindow[] { + if (windows.length <= 1) return windows; + + const snap = (t: Date) => { + if ( + t.getTime() === startDate.getTime() || + t.getTime() === endDate.getTime() + ) { + return t; + } + return new Date( + Math.max(Math.floor(t.getTime() / 1000) * 1000, startDate.getTime()), + ); + }; + + return windows + .map(w => ({ + ...w, + startTime: snap(w.startTime), + endTime: snap(w.endTime), + })) + .filter(w => w.startTime.getTime() < w.endTime.getTime()) + .map((w, windowIndex) => ({ ...w, windowIndex })); +} diff --git a/packages/app/src/hooks/__tests__/useChartConfig.test.tsx b/packages/app/src/hooks/__tests__/useChartConfig.test.tsx index a20788a176..d2cc3a3e72 100644 --- a/packages/app/src/hooks/__tests__/useChartConfig.test.tsx +++ b/packages/app/src/hooks/__tests__/useChartConfig.test.tsx @@ -27,8 +27,10 @@ import { import { useMVOptimizationExplanation } from '@/hooks/useMVOptimizationExplanation'; // Mock DEFAULT_TIME_WINDOWS_SECONDS to remove the 15m window -jest.mock('@/utils/searchWindows', () => { - const original = jest.requireActual('@/utils/searchWindows'); +jest.mock('@hyperdx/common-utils/dist/core/searchWindows', () => { + const original = jest.requireActual( + '@hyperdx/common-utils/dist/core/searchWindows', + ); const mockWindows = [ 6 * 60 * 60, // 6h 6 * 60 * 60, // 6h diff --git a/packages/app/src/hooks/useChartConfig.tsx b/packages/app/src/hooks/useChartConfig.tsx index ed367f0c27..fd30ab4597 100644 --- a/packages/app/src/hooks/useChartConfig.tsx +++ b/packages/app/src/hooks/useChartConfig.tsx @@ -16,6 +16,7 @@ import { isUsingGranularity, renderChartConfig, } from '@hyperdx/common-utils/dist/core/renderChartConfig'; +import { generateTimeWindowsDescending } from '@hyperdx/common-utils/dist/core/searchWindows'; import { convertDateRangeToGranularityString, convertGranularityToSeconds, @@ -55,7 +56,6 @@ import { queryPromqlChartConfig, reduceBucketRows, } from '@/utils/promqlChartQuery'; -import { generateTimeWindowsDescending } from '@/utils/searchWindows'; import { useMVOptimizationExplanation } from './useMVOptimizationExplanation'; diff --git a/packages/app/src/hooks/useOffsetPaginatedQuery.tsx b/packages/app/src/hooks/useOffsetPaginatedQuery.tsx index e8b5b573f6..5e08f9910d 100644 --- a/packages/app/src/hooks/useOffsetPaginatedQuery.tsx +++ b/packages/app/src/hooks/useOffsetPaginatedQuery.tsx @@ -14,6 +14,13 @@ import { } from '@hyperdx/common-utils/dist/clickhouse'; import { Metadata } from '@hyperdx/common-utils/dist/core/metadata'; import { renderChartConfig } from '@hyperdx/common-utils/dist/core/renderChartConfig'; +import { + DEFAULT_TIME_WINDOWS_SECONDS, + generateTimeWindowsAscending, + generateTimeWindowsDescending, + ONE_MIN_WINDOW, + TimeWindow, +} from '@hyperdx/common-utils/dist/core/searchWindows'; import { isFirstOrderByAscending, isTimestampExpressionInFirstOrderBy, @@ -42,13 +49,6 @@ import { useMVOptimizationExplanation } from '@/hooks/useMVOptimizationExplanati import { useQueryAttribution } from '@/queryAttribution'; import { useSource } from '@/source'; import { queryPromqlChartConfig } from '@/utils/promqlChartQuery'; -import { - DEFAULT_TIME_WINDOWS_SECONDS, - generateTimeWindowsAscending, - generateTimeWindowsDescending, - ONE_MIN_WINDOW, - TimeWindow, -} from '@/utils/searchWindows'; type TQueryKey = readonly [ string, diff --git a/packages/common-utils/src/__tests__/searchWindows.test.ts b/packages/common-utils/src/__tests__/searchWindows.test.ts new file mode 100644 index 0000000000..da5b35c064 --- /dev/null +++ b/packages/common-utils/src/__tests__/searchWindows.test.ts @@ -0,0 +1,75 @@ +import { + generateTimeWindowsAscending, + generateTimeWindowsDescending, + windowInclusivity, +} from '@/core/searchWindows'; + +const START = new Date('2026-05-10T00:00:00.000Z'); +const END = new Date('2026-05-11T00:00:00.000Z'); + +describe('windowInclusivity', () => { + it('makes descending windows [start, end), newest window owning the range end', () => { + const windows = generateTimeWindowsDescending(START, END); + expect(windows.length).toBeGreaterThan(1); + + expect(windowInclusivity(windows[0], windows.length)).toEqual({ + dateRangeStartInclusive: true, + dateRangeEndInclusive: true, + }); + expect( + windowInclusivity(windows[windows.length - 1], windows.length), + ).toEqual({ + dateRangeStartInclusive: true, + dateRangeEndInclusive: false, + }); + }); + + it('makes ascending windows half-open, newest window owning the range end', () => { + const windows = generateTimeWindowsAscending(START, END); + expect(windows.length).toBeGreaterThan(1); + + expect(windowInclusivity(windows[0], windows.length)).toEqual({ + dateRangeStartInclusive: true, + dateRangeEndInclusive: false, + }); + expect( + windowInclusivity(windows[windows.length - 1], windows.length), + ).toEqual({ + dateRangeStartInclusive: true, + dateRangeEndInclusive: true, + }); + }); + + it('leaves a single window fully inclusive on both ends', () => { + const windows = generateTimeWindowsDescending(START, START); + expect(windows).toHaveLength(1); + expect(windowInclusivity(windows[0], 1)).toEqual({ + dateRangeStartInclusive: true, + dateRangeEndInclusive: true, + }); + }); + + it.each([ + ['descending', generateTimeWindowsDescending(START, END)], + ['ascending', generateTimeWindowsAscending(START, END)], + ])( + 'lets exactly one %s window claim each shared boundary instant', + (_, windows) => { + // Every window includes its start; only the one holding the range end + // includes its end. + const ownsRangeEnd = (w: (typeof windows)[number]) => + w.endTime.getTime() === END.getTime(); + const wrongClaimants = windows.filter( + w => + !ownsRangeEnd(w) && + windowInclusivity(w, windows.length).dateRangeEndInclusive, + ); + expect(wrongClaimants).toEqual([]); + expect( + windows.every( + w => windowInclusivity(w, windows.length).dateRangeStartInclusive, + ), + ).toBe(true); + }, + ); +}); diff --git a/packages/app/src/utils/searchWindows.ts b/packages/common-utils/src/core/searchWindows.ts similarity index 75% rename from packages/app/src/utils/searchWindows.ts rename to packages/common-utils/src/core/searchWindows.ts index 4dd4fa61f7..57c6dce7cf 100644 --- a/packages/app/src/utils/searchWindows.ts +++ b/packages/common-utils/src/core/searchWindows.ts @@ -37,7 +37,7 @@ export function generateTimeWindowsDescending( while (currentEnd > startDate) { const windowSizeSeconds = - windowDurationsSeconds[windowIndex] || + windowDurationsSeconds.at(windowIndex) ?? windowDurationsSeconds[windowDurationsSeconds.length - 1]; // use largest window size const windowSizeMs = windowSizeSeconds * 1000; const windowStart = new Date( @@ -81,7 +81,7 @@ export function generateTimeWindowsAscending( while (currentStart < endDate) { const windowSizeSeconds = - windowDurationsSeconds[windowIndex] || + windowDurationsSeconds.at(windowIndex) ?? windowDurationsSeconds[windowDurationsSeconds.length - 1]; // use largest window size const windowSizeMs = windowSizeSeconds * 1000; const windowEnd = new Date( @@ -101,3 +101,28 @@ export function generateTimeWindowsAscending( return windows; } + +/** + * Windows are [start, end), so a row on a shared boundary lands on one page. + * The window holding the caller's endTime also includes its end. + */ +export function windowInclusivity( + window: TimeWindow, + windowCount: number, +): { dateRangeStartInclusive: boolean; dateRangeEndInclusive: boolean } { + const isLast = window.windowIndex === windowCount - 1; + + if (window.direction === 'DESC') { + // Windows walk backwards: index 0 is newest, the last index is oldest. + return { + dateRangeStartInclusive: true, + dateRangeEndInclusive: window.windowIndex === 0, + }; + } + + // ASC: index 0 is oldest, the last index is newest. + return { + dateRangeStartInclusive: true, + dateRangeEndInclusive: isLast, + }; +} From a4764398dfecf5dc29ab7a4fb3578fb2c2008102 Mon Sep 17 00:00:00 2001 From: vinzee Date: Sun, 25 Oct 2026 19:50:45 -0700 Subject: [PATCH 2/3] perf(api): sign search cursors and cut windows at the last row's second Cursors are HMAC-signed with EXPRESS_SESSION_SECRET and carry the current window's bounds, so the next window is computed directly and the range cap is gone. Each full page cuts its window at the last row's second, keeping OFFSET small in dense ranges. Only walks led by an exact timestamp key use windows; others read one window with the 10,000 offset cap. Cursor walks force the overflow modes to throw instead of truncating. --- .changeset/search-cursor-pagination.md | 8 +- packages/api/openapi.json | 2 +- .../external-api/__tests__/search.int.test.ts | 306 +++++++++++++++++- .../api/src/routers/external-api/v2/search.ts | 30 +- .../v2/utils/__tests__/searchCursor.test.ts | 122 ++++--- .../utils/__tests__/searchWindowCut.test.ts | 75 +++++ .../v2/utils/__tests__/searchWindows.test.ts | 243 +++++++++++++- .../routers/external-api/v2/utils/search.ts | 165 +++++++--- .../external-api/v2/utils/searchCursor.ts | 87 +++-- .../external-api/v2/utils/searchWindows.ts | 225 +++++++++---- .../src/__tests__/searchWindows.test.ts | 75 ----- .../common-utils/src/__tests__/utils.test.ts | 17 + .../src/core/renderChartConfig.ts | 10 +- .../src/core/searchChartConfig.ts | 2 +- .../common-utils/src/core/searchWindows.ts | 25 -- packages/common-utils/src/core/utils.ts | 17 + 16 files changed, 1098 insertions(+), 311 deletions(-) create mode 100644 packages/api/src/routers/external-api/v2/utils/__tests__/searchWindowCut.test.ts delete mode 100644 packages/common-utils/src/__tests__/searchWindows.test.ts diff --git a/.changeset/search-cursor-pagination.md b/.changeset/search-cursor-pagination.md index 6c0b92073a..2ccaae4af8 100644 --- a/.changeset/search-cursor-pagination.md +++ b/.changeset/search-cursor-pagination.md @@ -3,10 +3,4 @@ '@hyperdx/common-utils': patch --- -Add cursor pagination to the external API v2 search endpoint. Pass the -`nextCursor` from a response back as `cursor` to fetch the next page; pages are -scoped to a time window instead of a growing `OFFSET`, so page cost no longer -grows with depth. The existing `offset` parameter is unchanged. - -Keep requesting pages until `nextCursor` is `null` — a short page does not mean -the results are exhausted. +Add cursor pagination to the external API v2 search endpoint: send `cursor: "start"`, then pass back each `nextCursor` until it is `null`. Cursors are signed with `EXPRESS_SESSION_SECRET`: replicas need a shared value, and rotating it invalidates outstanding cursors. diff --git a/packages/api/openapi.json b/packages/api/openapi.json index ab147e52d3..9036893bc9 100644 --- a/packages/api/openapi.json +++ b/packages/api/openapi.json @@ -4818,7 +4818,7 @@ "type": "string", "maxLength": 1024, "example": "start", - "description": "Send the literal string \"start\" to begin cursor paging, then pass\nback the nextCursor from each response. When set, offset is\nignored; omit cursor entirely to use offset-based paging.\nCursor paging supports time ranges of up to 366 days. As with\noffset, rows that share a timestamp can be skipped or repeated\nbetween pages unless orderBy breaks the tie.\n\nKeep requesting pages until nextCursor is null. A page that is\nempty or returns fewer rows than maxResults does NOT mean the\nresults are exhausted.\n" + "description": "Send the literal string \"start\" to begin cursor paging, then pass\nback the nextCursor from each response. When set, offset is\nignored; omit cursor entirely to use offset-based paging.\nAs with offset, rows that share a timestamp can be skipped or\nrepeated between pages unless orderBy breaks the tie.\n\nKeep requesting pages until nextCursor is null. A page that is\nempty or returns fewer rows than maxResults does NOT mean the\nresults are exhausted.\n" } } }, diff --git a/packages/api/src/routers/external-api/__tests__/search.int.test.ts b/packages/api/src/routers/external-api/__tests__/search.int.test.ts index 4864f5bec0..3e7c836326 100644 --- a/packages/api/src/routers/external-api/__tests__/search.int.test.ts +++ b/packages/api/src/routers/external-api/__tests__/search.int.test.ts @@ -1,3 +1,4 @@ +import type { QuerySettings } from '@hyperdx/common-utils/dist/types'; import { SourceKind } from '@hyperdx/common-utils/dist/types'; import mongoose from 'mongoose'; import request from 'supertest'; @@ -18,6 +19,7 @@ import { encodeSearchCursor, fingerprintQuery, } from '@/routers/external-api/v2/utils/searchCursor'; +import logger from '@/utils/logger'; const DEFAULT_END_TIME = Math.floor(Date.now() / 60000) * 60000; const DEFAULT_START_TIME = DEFAULT_END_TIME - 3600 * 1000; @@ -582,6 +584,9 @@ describe('External API v2 Search', () => { ]); }; + const cursorOffset = (cursor: string) => + JSON.parse(Buffer.from(cursor.split('.')[0], 'base64url').toString()).o; + const cursorBody = () => ({ sourceId: logSource.id.toString(), startTime: iso(DEFAULT_START_TIME), @@ -703,21 +708,316 @@ describe('External API v2 Search', () => { }, ); - it('rejects a cursor walk over a range longer than a year', async () => { - await search({ + it('starts a cursor walk over a range longer than a year', async () => { + await seedCursorRows(); + + const res = await search({ ...cursorBody(), startTime: iso(DEFAULT_END_TIME - 400 * 24 * 60 * 60 * 1000), cursor: 'start', + }).expect(200); + expect(res.body.data).toHaveLength(5); + expect(res.body.nextCursor).toBeTruthy(); + }); + + it('pages past 10,000 rows in a range shorter than the first window', async () => { + const rangeStart = DEFAULT_END_TIME - 10 * 60 * 1000; + await bulkInsertLogs( + Array.from({ length: 10_050 }, (_, i) => ({ + ServiceName: 'cursor-dense-test', + Timestamp: new Date(rangeStart + i * 50), + SeverityText: 'INFO', + Body: `dense-${i}`, + })), + ); + const body = { + ...cursorBody(), + startTime: iso(rangeStart), + where: 'ServiceName:"cursor-dense-test"', + maxResults: 2000, + }; + + const seen = new Set(); + const offsets: number[] = []; + let cursor: string | null | undefined; + for (let pages = 0; pages < 20; pages++) { + const res = await search({ ...body, cursor: cursor ?? 'start' }).expect( + 200, + ); + res.body.data.forEach((r: Record) => seen.add(r.Body)); + cursor = res.body.nextCursor; + if (cursor == null) break; + offsets.push(cursorOffset(cursor)); + } + expect(cursor).toBeNull(); + expect(seen.size).toBe(10_050); + // Rows are 50ms apart, so a second holds 20 of them. + expect(Math.max(...offsets)).toBeLessThanOrEqual(20); + }); + + it.each(['Timestamp DESC', 'Timestamp ASC'])( + 'returns every row once when one second holds more rows than a page (%s)', + async orderBy => { + const second = DEFAULT_END_TIME - 5 * 60 * 1000; + await bulkInsertLogs( + Array.from({ length: 900 }, (_, i) => ({ + ServiceName: 'cursor-burst-test', + Timestamp: new Date(second + i), + SeverityText: 'INFO', + Body: `burst-${i}`, + })), + ); + const body = { + ...cursorBody(), + where: 'ServiceName:"cursor-burst-test"', + orderBy, + maxResults: 300, + }; + + const seen: string[] = []; + let cursor: string | null | undefined; + for (let pages = 0; pages < 20; pages++) { + const res = await search({ ...body, cursor: cursor ?? 'start' }).expect( + 200, + ); + seen.push(...res.body.data.map((r: Record) => r.Body)); + cursor = res.body.nextCursor; + if (cursor == null) break; + } + expect(cursor).toBeNull(); + expect(seen).toHaveLength(900); + expect(new Set(seen).size).toBe(900); + }, + ); + + it('walks every row exactly once in ascending order', async () => { + await seedCursorRows(); + // seedCursorRows puts four rows on one Timestamp; Body breaks the tie. + + const seen: string[] = []; + let cursor: string | null | undefined; + for (let pages = 0; pages < 50; pages++) { + const res = await search({ + ...cursorBody(), + orderBy: 'Timestamp ASC, Body ASC', + cursor: cursor ?? 'start', + }).expect(200); + seen.push(...res.body.data.map((r: Record) => r.Body)); + cursor = res.body.nextCursor; + if (cursor == null) break; + } + expect(cursor).toBeNull(); + expect(seen).toHaveLength(43); + expect(new Set(seen).size).toBe(43); + }); + + it.each([ + ['omitted', {}], + ['without a timestamp', { select: 'Body' }], + ])( + 'never returns the cursor timestamp column when select is %s', + async (_label, selectField) => { + await seedCursorRows(); + const { select: _omit, ...rest } = cursorBody(); + const res = await search({ + ...rest, + ...selectField, + cursor: 'start', + }).expect(200); + expect(res.body.data.length).toBeGreaterThan(0); + for (const row of res.body.data) { + expect(row).not.toHaveProperty('__hdx_cursor_ts'); + } + }, + ); + + const createLimitedSource = (querySettings: QuerySettings) => + Source.create({ + kind: SourceKind.Log, + team: logSource.team, + from: logSource.from, + defaultTableSelectExpression: 'Timestamp, SeverityText, Body', + timestampValueExpression: 'Timestamp', + connection: logSource.connection, + name: 'Logs (limited)', + querySettings, + }); + + it('fails a cursor walk instead of truncating when source settings say break', async () => { + await seedCursorRows(); + const limited = await createLimitedSource([ + { setting: 'max_rows_to_read', value: '1' }, + { setting: 'read_overflow_mode', value: 'break' }, + ]); + const errorSpy = jest.spyOn(logger, 'error'); + + const { maxResults: _max, ...body } = cursorBody(); + // Offset paging keeps the source's 'break' setting; a cursor walk overrides + // it and fails instead of truncating. + await search({ ...body, sourceId: limited.id.toString() }).expect(200); + expect(errorSpy).not.toHaveBeenCalled(); + const res = await search({ + ...body, + sourceId: limited.id.toString(), + cursor: 'start', + }); + expect(res.status).toBe(500); + expect(errorSpy).toHaveBeenCalledWith( + expect.objectContaining({ chType: 'TOO_MANY_ROWS' }), + '[search] ClickHouse query error', + ); + errorSpy.mockRestore(); + }); + + it.each([5, 100])( + 'walks every row when maxResults %i is at or above the source max_result_rows', + async maxResults => { + await seedCursorRows(); + const limited = await createLimitedSource([ + { setting: 'max_result_rows', value: '5' }, + { setting: 'result_overflow_mode', value: 'break' }, + ]); + + const seen: string[] = []; + let cursor: string | null | undefined; + for (let pages = 0; pages < 50; pages++) { + const res = await search({ + ...cursorBody(), + sourceId: limited.id.toString(), + maxResults, + cursor: cursor ?? 'start', + }).expect(200); + expect(res.body.data.length).toBeLessThanOrEqual(5); + seen.push(...res.body.data.map((r: Record) => r.Body)); + cursor = res.body.nextCursor; + if (cursor == null) break; + } + + expect(cursor).toBeNull(); + expect(seen).toHaveLength(43); + expect(new Set(seen).size).toBe(43); + }, + ); + + it('keeps a DISTINCT select distinct during a cursor walk', async () => { + const now = DEFAULT_END_TIME - 2 * 60 * 1000; + await bulkInsertLogs( + Array.from({ length: 30 }, (_, i) => ({ + ServiceName: 'cursor-distinct-test', + // Half the rows are older than the first 15m window's start. + Timestamp: new Date(now - (i < 15 ? i : 30 * 60 + i) * 1000), + SeverityText: 'INFO', + Body: `dup-${i % 3}`, + })), + ); + + const seen: string[] = []; + let cursor: string | null | undefined; + for (let pages = 0; pages < 20; pages++) { + const res = await search({ + ...cursorBody(), + where: 'ServiceName:"cursor-distinct-test"', + select: 'DISTINCT Body', + maxResults: 2, + cursor: cursor ?? 'start', + }).expect(200); + seen.push(...res.body.data.map((r: Record) => r.Body)); + cursor = res.body.nextCursor; + if (cursor == null) break; + } + expect(seen.sort()).toEqual(['dup-0', 'dup-1', 'dup-2']); + }); + + it.each([ + [ + 'a rounded leading key', + 'toStartOfMinute(Timestamp), Timestamp', + 'toStartOfMinute(Timestamp) DESC, Body ASC', + ], + [ + 'an exact leading key on a compound expression', + 'toDateTime(Timestamp), Timestamp', + 'Timestamp DESC', + ], + ])( + 'returns every row once when the walk is ordered by %s', + async (_label, timestampValueExpression, orderBy) => { + const compound = await Source.create({ + kind: SourceKind.Log, + team: logSource.team, + from: logSource.from, + defaultTableSelectExpression: 'Timestamp, SeverityText, Body', + timestampValueExpression, + connection: logSource.connection, + name: 'Logs (compound)', + }); + const minute = + Math.floor((DEFAULT_END_TIME - 3 * 60 * 1000) / 60_000) * 60_000; + // Body order is unrelated to time, so a cut on the wrong key loses rows. + await bulkInsertLogs( + Array.from({ length: 120 }, (_, i) => ({ + ServiceName: 'cursor-order-test', + Timestamp: new Date(minute + i * 400), + SeverityText: 'INFO', + Body: `row-${String((i * 37) % 120).padStart(3, '0')}`, + })), + ); + + const seen: string[] = []; + let cursor: string | null | undefined; + for (let pages = 0; pages < 50; pages++) { + const res = await search({ + ...cursorBody(), + sourceId: compound.id.toString(), + where: 'ServiceName:"cursor-order-test"', + orderBy, + maxResults: 10, + cursor: cursor ?? 'start', + }).expect(200); + seen.push(...res.body.data.map((r: Record) => r.Body)); + cursor = res.body.nextCursor; + if (cursor == null) break; + } + expect(cursor).toBeNull(); + expect(seen).toHaveLength(120); + expect(new Set(seen).size).toBe(120); + }, + ); + + it('rejects a select that uses the reserved cursor column name', async () => { + await search({ + ...cursorBody(), + select: 'Timestamp AS __hdx_cursor_ts', + cursor: 'start', }).expect(400); }); + it('rejects a cursor whose contents were edited', async () => { + await seedCursorRows(); + + const first = await search({ ...cursorBody(), cursor: 'start' }).expect( + 200, + ); + const [payload, signature] = String(first.body.nextCursor).split('.'); + const edited = JSON.parse(Buffer.from(payload, 'base64url').toString()); + edited.o += 1; + const forged = `${Buffer.from(JSON.stringify(edited)).toString('base64url')}.${signature}`; + + const res = await search({ ...cursorBody(), cursor: forged }).expect(400); + expect(res.body.message).toMatch(/invalid or expired/); + }); + const forgeCursor = ( body: ReturnType & { orderBy: string }, offset: number, ) => encodeSearchCursor( { - windowIndex: 0, + window: { + ordinal: 0, + startTime: iso(DEFAULT_START_TIME), + endTime: iso(DEFAULT_END_TIME), + }, offset, startTime: iso(DEFAULT_START_TIME), endTime: iso(DEFAULT_END_TIME), diff --git a/packages/api/src/routers/external-api/v2/search.ts b/packages/api/src/routers/external-api/v2/search.ts index 417d2cb4f8..dfba2068eb 100644 --- a/packages/api/src/routers/external-api/v2/search.ts +++ b/packages/api/src/routers/external-api/v2/search.ts @@ -21,12 +21,12 @@ import { externalDashboardSearchRequestSchema } from '@/utils/zod'; import { runSearchConfig, type SearchErrorCode } from './utils/search'; import { CURSOR_START, + CURSOR_TS_COLUMN, decodeSearchCursor, encodeSearchCursor, fingerprintQuery, - MAX_CURSOR_RANGE_MS, MAX_SINGLE_WINDOW_OFFSET, - type SearchCursorState, + type SearchWalk, } from './utils/searchCursor'; const searchQueryDuration = getHistogram('hyperdx.search.query.duration_ms', { @@ -150,9 +150,8 @@ const CH_USER_INPUT_ERRORS = new Set([ * Send the literal string "start" to begin cursor paging, then pass * back the nextCursor from each response. When set, offset is * ignored; omit cursor entirely to use offset-based paging. - * Cursor paging supports time ranges of up to 366 days. As with - * offset, rows that share a timestamp can be skipped or repeated - * between pages unless orderBy breaks the tie. + * As with offset, rows that share a timestamp can be skipped or + * repeated between pages unless orderBy breaks the tie. * * Keep requesting pages until nextCursor is null. A page that is * empty or returns fewer rows than maxResults does NOT mean the @@ -241,6 +240,9 @@ const searchRequestSchema = z.object({ 'select must not contain semicolons or SELECT subqueries; ' + 'use column references, map lookups, or scalar functions only', }) + .refine(value => !value.includes(CURSOR_TS_COLUMN), { + message: `select must not use the reserved name ${CURSOR_TS_COLUMN}`, + }) .describe( 'Comma-separated list of column expressions to return. ' + 'When omitted the source default columns are used. ' + @@ -397,7 +399,6 @@ function codeToStatus(code: SearchErrorCode): number { case 'SOURCE_NOT_FOUND': case 'CONNECTION_NOT_FOUND': return 404; - case 'CURSOR_QUERY_MISMATCH': case 'CURSOR_WINDOW_TOO_DENSE': return 400; default: { @@ -445,10 +446,10 @@ router.post( // The cursor pins this request's resolved range, so a defaulted endTime // stays the same on later pages. - let cursorState: SearchCursorState | undefined; + let cursorState: SearchWalk | undefined; if (cursor === CURSOR_START) { cursorState = { - windowIndex: 0, + window: null, offset: 0, startTime: startDate.toISOString(), endTime: endDate.toISOString(), @@ -461,7 +462,7 @@ router.post( message: decoded.error === 'CURSOR_QUERY_MISMATCH' ? 'Cursor does not match this query. Start a new search.' - : 'Cursor is malformed.', + : 'Cursor is invalid or expired. Start a new search with cursor "start".', }); } // Reject an explicitly changed bound; omitted or empty bounds default @@ -477,17 +478,6 @@ router.post( } cursorState = decoded; } - if ( - cursorState && - Date.parse(cursorState.endTime) - Date.parse(cursorState.startTime) > - MAX_CURSOR_RANGE_MS - ) { - searchQueryErrors.add(1, { error_type: 'CURSOR_RANGE_TOO_LARGE' }); - return res.status(400).json({ - message: - 'Cursor paging supports time ranges up to 366 days. Narrow startTime/endTime.', - }); - } const config = externalDashboardSearchRequestSchema.parse({ displayType: 'search' as const, diff --git a/packages/api/src/routers/external-api/v2/utils/__tests__/searchCursor.test.ts b/packages/api/src/routers/external-api/v2/utils/__tests__/searchCursor.test.ts index eb04787738..9b1be5fdbd 100644 --- a/packages/api/src/routers/external-api/v2/utils/__tests__/searchCursor.test.ts +++ b/packages/api/src/routers/external-api/v2/utils/__tests__/searchCursor.test.ts @@ -2,7 +2,7 @@ import { decodeSearchCursor, encodeSearchCursor, fingerprintQuery, - MAX_CURSOR_OFFSET, + type SearchCursorState, } from '@/routers/external-api/v2/utils/searchCursor'; const FP_INPUT = { @@ -18,26 +18,44 @@ const RANGE = { endTime: '2026-05-10T01:00:00.000Z', }; +const state = ( + overrides: Partial = {}, +): SearchCursorState => ({ + window: { + ordinal: 1, + startTime: '2026-05-10T00:30:00.000Z', + endTime: '2026-05-10T00:45:00.000Z', + }, + offset: 500, + ...RANGE, + ...overrides, +}); + +// Rewrites the payload half of a cursor while keeping its signature. +const tamper = (cursor: string, edit: (p: Record) => void) => { + const [payload, sig] = cursor.split('.'); + const parsed = JSON.parse(Buffer.from(payload, 'base64url').toString('utf8')); + edit(parsed); + return `${Buffer.from(JSON.stringify(parsed)).toString('base64url')}.${sig}`; +}; + describe('searchCursor', () => { - it('round-trips window index and offset', () => { + it('round-trips the window, offset and range', () => { const fp = fingerprintQuery(FP_INPUT); - const cursor = encodeSearchCursor( - { windowIndex: 3, offset: 500, ...RANGE }, - fp, + const cursor = encodeSearchCursor(state(), fp); + expect(decodeSearchCursor(cursor, fp)).toEqual(state()); + }); + + it('encodes the same state to the same cursor', () => { + const fp = fingerprintQuery(FP_INPUT); + expect(encodeSearchCursor(state(), fp)).toBe( + encodeSearchCursor(state(), fp), ); - expect(decodeSearchCursor(cursor, fp)).toEqual({ - windowIndex: 3, - offset: 500, - ...RANGE, - }); }); it('rejects a cursor minted for a different query', () => { const fp = fingerprintQuery(FP_INPUT); - const cursor = encodeSearchCursor( - { windowIndex: 1, offset: 0, ...RANGE }, - fp, - ); + const cursor = encodeSearchCursor(state(), fp); const otherFp = fingerprintQuery({ ...FP_INPUT, where: 'SeverityText:WARN', @@ -47,52 +65,66 @@ describe('searchCursor', () => { }); }); - it('rejects malformed and unknown-version cursors', () => { + it.each([ + ['the offset', (p: Record) => (p.o = 999_999)], + [ + 'the window start', + (p: Record) => (p.ws = RANGE.startTime), + ], + [ + 'the range end', + (p: Record) => (p.e = '2030-01-01T00:00:00.000Z'), + ], + ])('rejects a cursor whose %s was edited', (_, edit) => { const fp = fingerprintQuery(FP_INPUT); - expect(decodeSearchCursor('not-base64!!', fp)).toEqual({ + const edited = tamper(encodeSearchCursor(state(), fp), edit); + expect(decodeSearchCursor(edited, fp)).toEqual({ error: 'INVALID_CURSOR' }); + }); + + it('rejects a cursor signed with a different secret', () => { + const fp = fingerprintQuery(FP_INPUT); + const cursor = encodeSearchCursor(state(), fp, 'secret-a'); + expect(decodeSearchCursor(cursor, fp, 'secret-b')).toEqual({ error: 'INVALID_CURSOR', }); - const future = Buffer.from( - JSON.stringify({ v: 99, w: 0, o: 0, f: fp }), - ).toString('base64url'); - expect(decodeSearchCursor(future, fp)).toEqual({ error: 'INVALID_CURSOR' }); + expect(decodeSearchCursor(cursor, fp, 'secret-a')).toEqual(state()); }); - it('accepts in-window offsets past the request field cap', () => { - // A walk through a dense window passes the request field's 10,000 cap. + it('rejects malformed, unsigned and unknown-version cursors', () => { const fp = fingerprintQuery(FP_INPUT); - const deep = encodeSearchCursor( - { windowIndex: 0, offset: 50_000, ...RANGE }, - fp, - ); - expect(decodeSearchCursor(deep, fp)).toMatchObject({ offset: 50_000 }); + expect(decodeSearchCursor('not-base64!!', fp)).toEqual({ + error: 'INVALID_CURSOR', + }); + const [payload] = encodeSearchCursor(state(), fp).split('.'); + expect(decodeSearchCursor(payload, fp)).toEqual({ + error: 'INVALID_CURSOR', + }); + const future = tamper(encodeSearchCursor(state(), fp), p => (p.v = 99)); + expect(decodeSearchCursor(future, fp)).toEqual({ error: 'INVALID_CURSOR' }); }); - it('rejects an offset past the sanity bound', () => { + it('rejects a window outside the pinned range', () => { const fp = fingerprintQuery(FP_INPUT); - const over = encodeSearchCursor( - { windowIndex: 0, offset: MAX_CURSOR_OFFSET + 1, ...RANGE }, + const outside = encodeSearchCursor( + state({ + window: { + ordinal: 0, + startTime: '2026-05-10T00:45:00.000Z', + endTime: '2026-05-10T01:15:00.000Z', + }, + }), fp, ); - expect(decodeSearchCursor(over, fp)).toEqual({ error: 'INVALID_CURSOR' }); - - const atBound = encodeSearchCursor( - { windowIndex: 0, offset: MAX_CURSOR_OFFSET, ...RANGE }, - fp, - ); - expect(decodeSearchCursor(atBound, fp)).toMatchObject({ - offset: MAX_CURSOR_OFFSET, + expect(decodeSearchCursor(outside, fp)).toEqual({ + error: 'INVALID_CURSOR', }); }); - it('carries the time range instead of fingerprinting it', () => { - // A defaulted endTime differs on every request. + it('accepts in-window offsets past the request field cap', () => { + // A walk through a dense window passes the request field's 10,000 cap. const fp = fingerprintQuery(FP_INPUT); - const cursor = encodeSearchCursor( - { windowIndex: 1, offset: 5, ...RANGE }, - fp, - ); - expect(decodeSearchCursor(cursor, fp)).toMatchObject(RANGE); + const deep = encodeSearchCursor(state({ offset: 50_000 }), fp); + expect(decodeSearchCursor(deep, fp)).toMatchObject({ offset: 50_000 }); }); it('changes the fingerprint for every field that changes the result set', () => { diff --git a/packages/api/src/routers/external-api/v2/utils/__tests__/searchWindowCut.test.ts b/packages/api/src/routers/external-api/v2/utils/__tests__/searchWindowCut.test.ts new file mode 100644 index 0000000000..44f3311a84 --- /dev/null +++ b/packages/api/src/routers/external-api/v2/utils/__tests__/searchWindowCut.test.ts @@ -0,0 +1,75 @@ +import { cutWindowAtSecond } from '@/routers/external-api/v2/utils/searchWindows'; + +const window = { + ordinal: 0, + startTime: '2026-10-25T11:45:00.000Z', + endTime: '2026-10-25T12:00:00.400Z', +}; +const ms = (iso: string) => Date.parse(iso); + +describe('cutWindowAtSecond', () => { + it('ends a descending window just after the last row second', () => { + expect( + cutWindowAtSecond({ + window, + direction: 'DESC', + offset: 0, + pageLength: 2000, + lastSecondMs: ms('2026-10-25T11:58:31.000Z'), + trailingRunLength: 12, + }), + ).toEqual({ + window: { ...window, endTime: '2026-10-25T11:58:32.000Z' }, + offset: 12, + }); + }); + + it('keeps the range end when the last row second reaches it', () => { + expect( + cutWindowAtSecond({ + window, + direction: 'DESC', + offset: 0, + pageLength: 2000, + lastSecondMs: ms('2026-10-25T12:00:00.000Z'), + trailingRunLength: 2000, + }), + ).toEqual({ window, offset: 2000 }); + }); + + it('starts an ascending window at the last row second', () => { + expect( + cutWindowAtSecond({ + window, + direction: 'ASC', + offset: 0, + pageLength: 2000, + lastSecondMs: ms('2026-10-25T11:46:10.000Z'), + trailingRunLength: 7, + }), + ).toEqual({ + window: { ...window, startTime: '2026-10-25T11:46:10.000Z' }, + offset: 7, + }); + }); + + it.each(['DESC', 'ASC'] as const)( + 'adds the page to the offset when the whole %s page was one second', + direction => { + const cut = + direction === 'DESC' + ? { ...window, endTime: '2026-10-25T11:58:32.000Z' } + : { ...window, startTime: '2026-10-25T11:58:31.000Z' }; + expect( + cutWindowAtSecond({ + window: cut, + direction, + offset: 300, + pageLength: 1000, + lastSecondMs: ms('2026-10-25T11:58:31.000Z'), + trailingRunLength: 1000, + }), + ).toEqual({ window: cut, offset: 1300 }); + }, + ); +}); diff --git a/packages/api/src/routers/external-api/v2/utils/__tests__/searchWindows.test.ts b/packages/api/src/routers/external-api/v2/utils/__tests__/searchWindows.test.ts index 9817283c0f..7041738313 100644 --- a/packages/api/src/routers/external-api/v2/utils/__tests__/searchWindows.test.ts +++ b/packages/api/src/routers/external-api/v2/utils/__tests__/searchWindows.test.ts @@ -1,11 +1,78 @@ -import { resolveSearchWindows } from '@/routers/external-api/v2/utils/searchWindows'; +import { + DEFAULT_TIME_WINDOWS_SECONDS, + generateTimeWindowsAscending, + generateTimeWindowsDescending, +} from '@hyperdx/common-utils/dist/core/searchWindows'; + +import { + nextSearchWindow, + resolveFirstSearchWindow, + windowBounds, +} from '@/routers/external-api/v2/utils/searchWindows'; + +type Opts = Parameters[4]; + +// Every window of a walk, in the order pages visit them. +async function resolveSearchWindows( + source: { timestampValueExpression?: string }, + orderBy: string, + start: Date, + end: Date, + opts?: Opts, +) { + const first = await resolveFirstSearchWindow( + source, + orderBy, + start, + end, + opts, + ); + const range = { start, end }; + const windows: { + startTime: Date; + endTime: Date; + windowIndex: number; + direction: 'ASC' | 'DESC'; + }[] = []; + for ( + let w: typeof first.window | null = first.window; + w != null; + w = nextSearchWindow(w, range, first.direction) + ) { + windows.push({ + startTime: new Date(w.startTime), + endTime: new Date(w.endTime), + windowIndex: w.ordinal, + direction: first.direction, + }); + } + return windows; +} + +// Reference windows: build every window, floor the boundaries between +// them to the second, drop empties. Stepping must reproduce it exactly. +function referenceWindows(start: Date, end: Date, direction: 'ASC' | 'DESC') { + const windows = + direction === 'ASC' + ? generateTimeWindowsAscending(start, end) + : generateTimeWindowsDescending(start, end); + if (windows.length <= 1) return windows.map(w => [w.startTime, w.endTime]); + const snap = (t: Date) => + t.getTime() === start.getTime() || t.getTime() === end.getTime() + ? t.getTime() + : Math.max(Math.floor(t.getTime() / 1000) * 1000, start.getTime()); + return windows + .map(w => [snap(w.startTime), snap(w.endTime)]) + .filter(([a, b]) => a < b) + .map(([a, b]) => [new Date(a), new Date(b)]); +} const source = { timestampValueExpression: 'Timestamp' }; const START = new Date('2026-05-10T00:00:00.000Z'); const END = new Date('2026-05-11T00:00:00.000Z'); -describe('resolveSearchWindows', () => { +describe('search window stepping', () => { it('windows a timestamp-led descending order', async () => { const windows = await resolveSearchWindows( source, @@ -158,6 +225,178 @@ describe('resolveSearchWindows', () => { }, ); + it.each(['DESC', 'ASC'] as const)( + 'steps through exactly the reference windows (%s)', + async direction => { + // Deterministic pseudo-random ranges: 0 to 30 days, millisecond precision. + let seed = 42; + const rand = () => + (seed = (seed * 1_103_515_245 + 12_345) % 2 ** 31) / 2 ** 31; + // Plus edge ranges: a mid-second start, with lengths within a second of + // each cumulative window boundary, where flooring can empty a window. + const edgeStart = new Date('2026-05-10T00:00:00.700Z'); + let cumulative = 0; + for (const size of [...DEFAULT_TIME_WINDOWS_SECONDS, 86_400, 86_400]) { + cumulative += size * 1000; + for (const delta of [-1300, -300, 0, 300, 1300]) { + const end = new Date(edgeStart.getTime() + cumulative + delta); + const walked = await resolveSearchWindows( + source, + `Timestamp ${direction}`, + edgeStart, + end, + ); + expect(walked.map(w => [w.startTime, w.endTime])).toEqual( + referenceWindows(edgeStart, end, direction), + ); + } + } + for (let i = 0; i < 200; i++) { + const start = new Date( + Date.UTC(2026, 4, 1) + Math.floor(rand() * 86_400_000), + ); + const end = new Date( + start.getTime() + Math.floor(rand() * 30 * 86_400_000), + ); + const walked = await resolveSearchWindows( + source, + `Timestamp ${direction}`, + start, + end, + ); + expect(walked.map(w => [w.startTime, w.endTime])).toEqual( + referenceWindows(start, end, direction), + ); + } + }, + ); + + it('steps through windows sized by the default table, then 24h repeating', async () => { + const windows = await resolveSearchWindows( + source, + 'Timestamp DESC', + START, + new Date(START.getTime() + 3 * 86_400_000), + ); + const sizes = windows.map( + w => (w.endTime.getTime() - w.startTime.getTime()) / 1000, + ); + expect(sizes.slice(0, DEFAULT_TIME_WINDOWS_SECONDS.length)).toEqual( + DEFAULT_TIME_WINDOWS_SECONDS, + ); + expect( + sizes + .slice(DEFAULT_TIME_WINDOWS_SECONDS.length, -1) + .every(s => s === 86_400), + ).toBe(true); + }); + + it.each([ + ['EventDate, toDateTime(ts)', 'toDateTime(ts) DESC', 'toDateTime(ts)'], + [ + 'TimestampTime, Timestamp', + '(TimestampTime, Timestamp) DESC', + 'TimestampTime', + ], + ['TimestampTime, Timestamp', 'Timestamp DESC, Body', 'Timestamp'], + ['toStartOfMinute(Timestamp), Timestamp', 'Timestamp DESC', 'Timestamp'], + ])( + 'cuts %s ordered by %s on %s', + async (timestampValueExpression, orderBy, cutPart) => { + const first = await resolveFirstSearchWindow( + { timestampValueExpression }, + orderBy, + START, + END, + { + columnType: async column => + column === 'EventDate' + ? 'Date' + : column === 'TimestampTime' || column === 'EventTime' + ? 'DateTime' + : 'DateTime64(9)', + }, + ); + expect(first.windowed).toBe(true); + expect(first.cutPart).toBe(cutPart); + }, + ); + + it.each([ + [ + 'toStartOfMinute(Timestamp), Timestamp', + 'toStartOfMinute(Timestamp) DESC, Body', + ], + ['EventDate, EventTime', 'EventDate DESC, Body'], + ])( + 'uses one window for %s ordered by a rounded key (%s)', + async (timestampValueExpression, orderBy) => { + // A window boundary inside one rounded key would break the requested order. + const first = await resolveFirstSearchWindow( + { timestampValueExpression }, + orderBy, + START, + END, + { + columnType: async column => + column === 'EventDate' ? 'Date' : 'DateTime64(9)', + }, + ); + expect(first.windowed).toBe(false); + expect(first.cutPart).toBeNull(); + }, + ); + + it('reports windowing for a range shorter than the first window', async () => { + const short = await resolveFirstSearchWindow( + source, + 'Timestamp DESC', + START, + new Date(START.getTime() + 10 * 60 * 1000), + ); + expect(short.windowed).toBe(true); + const unordered = await resolveFirstSearchWindow( + source, + 'Body DESC', + START, + END, + ); + expect(unordered.windowed).toBe(false); + const offsetOnly = await resolveFirstSearchWindow( + source, + 'Timestamp DESC', + START, + END, + { + forceSingleWindow: true, + }, + ); + expect(offsetOnly.windowed).toBe(false); + }); + + it('includes a window end only when it is the range end', () => { + const range = { start: START, end: END }; + const newest = { + ordinal: 0, + startTime: '2026-05-10T23:45:00.000Z', + endTime: END.toISOString(), + }; + const older = { + ordinal: 1, + startTime: '2026-05-10T17:45:00.000Z', + endTime: '2026-05-10T23:45:00.000Z', + }; + expect(windowBounds(newest, range)).toEqual({ + dateRange: [new Date(newest.startTime), END], + dateRangeStartInclusive: true, + dateRangeEndInclusive: true, + }); + expect(windowBounds(older, range)).toMatchObject({ + dateRangeStartInclusive: true, + dateRangeEndInclusive: false, + }); + }); + it('windows a default tuple order led by the timestamp', async () => { const tupleSource = { timestampValueExpression: 'TimestampTime' }; const windows = await resolveSearchWindows( diff --git a/packages/api/src/routers/external-api/v2/utils/search.ts b/packages/api/src/routers/external-api/v2/utils/search.ts index 9d888bfb21..e0c5215f02 100644 --- a/packages/api/src/routers/external-api/v2/utils/search.ts +++ b/packages/api/src/routers/external-api/v2/utils/search.ts @@ -1,12 +1,15 @@ import { getMetadata } from '@hyperdx/common-utils/dist/core/metadata'; -import { buildSearchChartConfig } from '@hyperdx/common-utils/dist/core/searchChartConfig'; -import { windowInclusivity } from '@hyperdx/common-utils/dist/core/searchWindows'; +import { + buildSearchChartConfig, + resolveSelect, +} from '@hyperdx/common-utils/dist/core/searchChartConfig'; import { getFirstTimestampValueExpression, splitAndTrimWithBracket, } from '@hyperdx/common-utils/dist/core/utils'; import type { ChartConfigWithDateRange, + QuerySettings, TSource, } from '@hyperdx/common-utils/dist/types'; import { @@ -21,16 +24,53 @@ import { getSource } from '@/controllers/sources'; import type { ExternalDashboardSearchRequestConfig } from '@/utils/zod'; import { - MAX_CURSOR_OFFSET, + CURSOR_TS_COLUMN, MAX_SINGLE_WINDOW_OFFSET, type SearchCursorState, + type SearchWalk, } from './searchCursor'; -import { resolveSearchWindows } from './searchWindows'; +import { + cutWindowAtSecond, + nextSearchWindow, + resolveFirstSearchWindow, + windowBounds, +} from './searchWindows'; + +const OVERFLOW_MODE_SETTING = /_overflow_mode(_leaf)?$/; +const DEFAULT_OVERFLOW_MODE_SETTINGS = [ + 'result_overflow_mode', + 'read_overflow_mode', + 'timeout_overflow_mode', +]; + +// A limit set to 'break' returns a partial page with no error, which a walk +// would read as the end of its window. Cursor walks fail loudly instead. +function withThrowingOverflow( + querySettings: QuerySettings | undefined, +): QuerySettings { + const settings = querySettings ?? []; + const overflowModes = new Set([ + ...DEFAULT_OVERFLOW_MODE_SETTINGS, + ...settings + .map(({ setting }) => setting) + .filter(setting => OVERFLOW_MODE_SETTING.test(setting)), + ]); + return [ + ...settings.filter(({ setting }) => !overflowModes.has(setting)), + ...[...overflowModes].map(setting => ({ setting, value: 'throw' })), + ]; +} + +function maxResultRows(querySettings: QuerySettings | undefined): number { + const value = Number( + querySettings?.find(s => s.setting === 'max_result_rows')?.value, + ); + return value > 0 ? value : Infinity; +} export type SearchErrorCode = | 'SOURCE_NOT_FOUND' | 'CONNECTION_NOT_FOUND' - | 'CURSOR_QUERY_MISMATCH' | 'CURSOR_WINDOW_TOO_DENSE'; type SearchError = { @@ -106,7 +146,7 @@ export async function runSearchConfig({ endDate: Date; maxResults: number; offset: number; - cursorState?: SearchCursorState; + cursorState?: SearchWalk; }): Promise { const source = await getSource(teamId, config.sourceId); if (!source) { @@ -158,14 +198,19 @@ export async function runSearchConfig({ const metadata = getMetadata(clickhouseClient); - // Offset-only requests use one window over the whole range. - const windows = await resolveSearchWindows( + const baseSelect = resolveSelect(source, config.select); + // Windows are cut via a hidden select column, which would change what + // DISTINCT deduplicates; DISTINCT is also per window. Those walks, and + // offset-only requests, use one window over the whole range. + const canWindow = + typeof baseSelect === 'string' && !/^\s*DISTINCT\b/i.test(baseSelect); + const first = await resolveFirstSearchWindow( source, effectiveOrderBy, walkStart, walkEnd, { - forceSingleWindow: !cursorState, + forceSingleWindow: !cursorState || !canWindow, columnType: async column => ( await metadata.getColumn({ @@ -177,23 +222,27 @@ export async function runSearchConfig({ )?.type, }, ); - const windowIndex = cursorState?.windowIndex ?? 0; - const window = windows[windowIndex]; - if (window == null) { - return { - isError: true, - code: 'CURSOR_QUERY_MISMATCH', - message: 'Cursor refers to a page outside the current time range', - }; - } + const range = { start: walkStart, end: walkEnd }; + const window = cursorState?.window ?? first.window; const effectiveOffset = cursorState ? cursorState.offset : offset; - // A single window keeps the `offset` field's cap. Erroring, rather than - // ending the walk, keeps a truncated result from looking complete. - const offsetCap = - windows.length === 1 ? MAX_SINGLE_WINDOW_OFFSET : MAX_CURSOR_OFFSET; + const cutColumn = cursorState ? first.cutPart : null; + // A walk that can't cut its windows pages with plain offsets, so it keeps + // the `offset` field's cap. Erroring, rather than ending the walk, keeps a + // truncated result from looking complete. + const requestedLimit = + cursorState && !cutColumn + ? Math.min(maxResults, MAX_SINGLE_WINDOW_OFFSET - effectiveOffset) + : maxResults; + // A cursor walk reads one extra row to learn whether the window continues, + // unless the source's max_result_rows leaves no room for it; then a full + // page counts as "more in the window". + const rowCap = maxResultRows(source.querySettings); const pageLimit = cursorState - ? Math.min(maxResults, offsetCap - effectiveOffset) - : maxResults; + ? Math.min(requestedLimit, rowCap) + : requestedLimit; + const queryLimit = cursorState + ? Math.min(pageLimit + 1, rowCap) + : requestedLimit; if (cursorState && pageLimit <= 0) { return { isError: true, @@ -206,27 +255,26 @@ export async function runSearchConfig({ const searchBase = buildSearchChartConfig(source, { where: typeof config.where === 'string' ? config.where : '', whereLanguage: config.whereLanguage ?? 'lucene', - select: config.select ?? null, + select: cutColumn + ? `${baseSelect || '*'}, toUnixTimestamp64Milli(toDateTime64(${cutColumn}, 3)) AS ${CURSOR_TS_COLUMN}` + : (config.select ?? null), displayType: DisplayType.Search, orderBy: effectiveOrderBy, - dateRange: [window.startTime, window.endTime], - ...windowInclusivity(window, windows.length), + ...windowBounds(window, range), }); const chartConfig: ChartConfigWithDateRange = { ...searchBase, connection: source.connection.toString(), - // A cursor walk reads one extra row to learn whether the window continues. - limit: { - limit: cursorState ? pageLimit + 1 : maxResults, - offset: effectiveOffset, - }, + limit: { limit: queryLimit, offset: effectiveOffset }, } as ChartConfigWithDateRange; const result = await clickhouseClient.queryChartConfig({ config: chartConfig, metadata, - querySettings: source.querySettings, + querySettings: cursorState + ? withThrowingOverflow(source.querySettings) + : source.querySettings, }); if ( @@ -242,8 +290,22 @@ export async function runSearchConfig({ string, unknown >[]; - const hasMoreInWindow = cursorState != null && rows.length > pageLimit; - const data = cursorState ? rows.slice(0, pageLimit) : rows; + const hasMoreInWindow = + cursorState != null && + (rows.length > pageLimit || + (queryLimit === pageLimit && rows.length === pageLimit)); + const page = cursorState ? rows.slice(0, pageLimit) : rows; + const split = cutColumn + ? page.map(({ [CURSOR_TS_COLUMN]: ms, ...row }) => { + const second = Math.floor(Number(ms) / 1000) * 1000; + if (!Number.isFinite(second)) { + throw new Error(`Search result is missing ${CURSOR_TS_COLUMN}`); + } + return { second, row }; + }) + : null; + const seconds = split?.map(r => r.second) ?? []; + const data = split?.map(r => r.row) ?? page; // Only a cursor walk gets a next cursor; offset-only callers are unchanged. let nextCursorState: SearchCursorState | null = null; @@ -252,11 +314,34 @@ export async function runSearchConfig({ startTime: cursorState.startTime, endTime: cursorState.endTime, }; - const nextOffset = effectiveOffset + data.length; - if (hasMoreInWindow) { - nextCursorState = { windowIndex, offset: nextOffset, ...pinned }; - } else if (windowIndex + 1 < windows.length) { - nextCursorState = { windowIndex: windowIndex + 1, offset: 0, ...pinned }; + if (hasMoreInWindow && cutColumn) { + const lastSecond = seconds[seconds.length - 1]; + let trailingRunLength = 0; + while (seconds[seconds.length - 1 - trailingRunLength] === lastSecond) { + trailingRunLength++; + } + nextCursorState = { + ...cutWindowAtSecond({ + window, + direction: first.direction, + offset: effectiveOffset, + pageLength: data.length, + lastSecondMs: lastSecond, + trailingRunLength, + }), + ...pinned, + }; + } else if (hasMoreInWindow) { + nextCursorState = { + window, + offset: effectiveOffset + data.length, + ...pinned, + }; + } else { + const nextWindow = nextSearchWindow(window, range, first.direction); + if (nextWindow) { + nextCursorState = { window: nextWindow, offset: 0, ...pinned }; + } } } diff --git a/packages/api/src/routers/external-api/v2/utils/searchCursor.ts b/packages/api/src/routers/external-api/v2/utils/searchCursor.ts index eebcf1ef1d..26973af190 100644 --- a/packages/api/src/routers/external-api/v2/utils/searchCursor.ts +++ b/packages/api/src/routers/external-api/v2/utils/searchCursor.ts @@ -1,28 +1,37 @@ -import { createHash } from 'crypto'; +import { createHash, createHmac, timingSafeEqual } from 'crypto'; + +import { EXPRESS_SESSION_SECRET } from '@/config'; const CURSOR_VERSION = 1; const FINGERPRINT_LENGTH = 16; -// Above the `offset` field's 10,000 cap so a walk can page through a dense -// window; still bounded because cursors are unsigned. -export const MAX_CURSOR_OFFSET = 1_000_000; // Also the `offset` request field's cap. export const MAX_SINGLE_WINDOW_OFFSET = 10_000; - -// Windows are built up front, so the range is capped. Also checked on decoded -// cursors, which are unsigned. -export const MAX_CURSOR_RANGE_MS = 366 * 24 * 60 * 60 * 1000; +const SIGNATURE_CONTEXT = 'hdx-search-cursor:'; +/** Hidden column carrying each row's second; reserved in `select`. */ +export const CURSOR_TS_COLUMN = '__hdx_cursor_ts'; /** Literal `cursor` value that starts a windowed cursor walk. */ export const CURSOR_START = 'start'; +export type CursorWindow = { + ordinal: number; + startTime: string; + endTime: string; +}; + export type SearchCursorState = { - windowIndex: number; + window: CursorWindow; offset: number; /** Time range pinned by the walk, so a defaulted endTime stays stable. */ startTime: string; endTime: string; }; +/** A walk in progress; `window` is null on its first page. */ +export type SearchWalk = Omit & { + window: CursorWindow | null; +}; + export type CursorFingerprintInput = { sourceId: string; where: string; @@ -55,29 +64,52 @@ export function fingerprintQuery(input: CursorFingerprintInput): string { .slice(0, FINGERPRINT_LENGTH); } +// Signed with the session secret, so a cursor can only come from this server +// and is valid wherever its sessions are. +function sign(payload: string, secret: string): Buffer { + return createHmac('sha256', secret) + .update(SIGNATURE_CONTEXT + payload) + .digest(); +} + export function encodeSearchCursor( state: SearchCursorState, fingerprint: string, + secret: string = EXPRESS_SESSION_SECRET, ): string { - return Buffer.from( + const payload = Buffer.from( JSON.stringify({ v: CURSOR_VERSION, - w: state.windowIndex, + k: state.window.ordinal, + ws: state.window.startTime, + we: state.window.endTime, o: state.offset, f: fingerprint, s: state.startTime, e: state.endTime, }), ).toString('base64url'); + return `${payload}.${sign(payload, secret).toString('base64url')}`; } export function decodeSearchCursor( cursor: string, fingerprint: string, + secret: string = EXPRESS_SESSION_SECRET, ): SearchCursorState | CursorDecodeError { + const [payload, signature, ...rest] = cursor.split('.'); + if (!payload || !signature || rest.length > 0) { + return { error: 'INVALID_CURSOR' }; + } + const expected = sign(payload, secret); + const actual = Buffer.from(signature, 'base64url'); + if (actual.length !== expected.length || !timingSafeEqual(actual, expected)) { + return { error: 'INVALID_CURSOR' }; + } + let parsed: unknown; try { - parsed = JSON.parse(Buffer.from(cursor, 'base64url').toString('utf8')); + parsed = JSON.parse(Buffer.from(payload, 'base64url').toString('utf8')); } catch { return { error: 'INVALID_CURSOR' }; } @@ -86,26 +118,39 @@ export function decodeSearchCursor( return { error: 'INVALID_CURSOR' }; } const fields: Record = { ...parsed }; - const { v, w, o, f, s: start, e: end } = fields; + const { v, k, ws, we, o, f, s: start, e: end } = fields; if ( v !== CURSOR_VERSION || - typeof w !== 'number' || + typeof k !== 'number' || typeof o !== 'number' || - !Number.isInteger(w) || + !Number.isInteger(k) || !Number.isInteger(o) || - w < 0 || + k < 0 || o < 0 || - o > MAX_CURSOR_OFFSET || typeof f !== 'string' || + typeof ws !== 'string' || + typeof we !== 'string' || typeof start !== 'string' || - typeof end !== 'string' || - Number.isNaN(Date.parse(start)) || - Number.isNaN(Date.parse(end)) + typeof end !== 'string' + ) { + return { error: 'INVALID_CURSOR' }; + } + const [wsMs, weMs, startMs, endMs] = [ws, we, start, end].map(Date.parse); + if ( + [wsMs, weMs, startMs, endMs].some(Number.isNaN) || + wsMs < startMs || + weMs > endMs || + wsMs > weMs ) { return { error: 'INVALID_CURSOR' }; } if (f !== fingerprint) { return { error: 'CURSOR_QUERY_MISMATCH' }; } - return { windowIndex: w, offset: o, startTime: start, endTime: end }; + return { + window: { ordinal: k, startTime: ws, endTime: we }, + offset: o, + startTime: start, + endTime: end, + }; } diff --git a/packages/api/src/routers/external-api/v2/utils/searchWindows.ts b/packages/api/src/routers/external-api/v2/utils/searchWindows.ts index fc85fddf29..95b9a789c1 100644 --- a/packages/api/src/routers/external-api/v2/utils/searchWindows.ts +++ b/packages/api/src/routers/external-api/v2/utils/searchWindows.ts @@ -1,52 +1,48 @@ +import { DEFAULT_TIME_WINDOWS_SECONDS } from '@hyperdx/common-utils/dist/core/searchWindows'; import { - generateTimeWindowsAscending, - generateTimeWindowsDescending, - TimeWindow, -} from '@hyperdx/common-utils/dist/core/searchWindows'; -import { - classifyTimestampType, getFirstOrderingItem, isFirstOrderByAscending, - parseToStartOfFunction, + isRoundedTimestampPart, removeTrailingDirection, splitAndTrimWithBracket, } from '@hyperdx/common-utils/dist/core/utils'; +import type { CursorWindow } from './searchCursor'; + // Windows keep the requested order only when the first sort key is the // timestamp itself, not a derived key such as toHour(Timestamp). Stricter than // isTimestampExpressionInFirstOrderBy, which accepts any key containing it. -function ordersByTimestamp( +function leadingTimestampPart( orderBy: string, timestampValueExpression: string | undefined, -): boolean { +): string | null { const first = getFirstOrderingItem(orderBy); if (typeof first !== 'string' || timestampValueExpression == null) { - return false; + return null; } const expression = removeTrailingDirection(first); // A tuple order such as (TimestampTime, Timestamp) is led by its first element. const leading = /^\(.*\)$/s.test(expression) ? splitAndTrimWithBracket(expression.slice(1, -1))[0] : expression; - return splitAndTrimWithBracket(timestampValueExpression).includes(leading); + return splitAndTrimWithBracket(timestampValueExpression).includes(leading) + ? leading + : null; } // toStartOf*, toDate and Date-typed parts get inclusive bounds on both ends. -// Windows stay half-open only if another part compares exact times. -async function keepsExactTime( - timestampValueExpression: string, +async function isExactTimestampPart( + part: string, columnType: (column: string) => Promise, ): Promise { - for (const part of splitAndTrimWithBracket(timestampValueExpression)) { - if (parseToStartOfFunction(part) || /^toDate\s*\(/i.test(part)) continue; - if (classifyTimestampType(await columnType(part))?.kind === 'date') - continue; - return true; - } - return false; + if (isRoundedTimestampPart(part, undefined)) return false; + return !isRoundedTimestampPart(part, await columnType(part)); } -export async function resolveSearchWindows( +export type SearchDirection = 'ASC' | 'DESC'; +type Range = { start: Date; end: Date }; + +export async function resolveFirstSearchWindow( source: { timestampValueExpression?: string }, orderBy: string, startDate: Date, @@ -58,57 +54,156 @@ export async function resolveSearchWindows( forceSingleWindow?: boolean; columnType?: (column: string) => Promise; } = {}, -): Promise { - const canWindow = +): Promise<{ + window: CursorWindow; + windowed: boolean; + /** The exact timestamp part leading the sort; windows are cut on it. */ + cutPart: string | null; + direction: SearchDirection; +}> { + const direction = isFirstOrderByAscending(orderBy) ? 'ASC' : 'DESC'; + const wholeRange = { + ordinal: 0, + startTime: startDate.toISOString(), + endTime: endDate.toISOString(), + }; + const leading = leadingTimestampPart( + orderBy, + source.timestampValueExpression, + ); + // Windows need an exact leading key: a boundary inside one rounded key + // (a minute, a day) would split rows the sort keeps together. + const cutPart = !forceSingleWindow && - ordersByTimestamp(orderBy, source.timestampValueExpression) && - (await keepsExactTime(source.timestampValueExpression ?? '', columnType)); - - if (!canWindow) { - return [ - { - startTime: startDate, - endTime: endDate, - windowIndex: 0, - direction: isFirstOrderByAscending(orderBy) ? 'ASC' : 'DESC', - }, - ]; + startDate < endDate && + leading != null && + (await isExactTimestampPart(leading, columnType)) + ? leading + : null; + if (cutPart == null) { + return { window: wholeRange, windowed: false, cutPart, direction }; } - const windows = isFirstOrderByAscending(orderBy) - ? generateTimeWindowsAscending(startDate, endDate) - : generateTimeWindowsDescending(startDate, endDate); - return snapBoundariesToSeconds(windows, startDate, endDate); + // Step from an empty window at the edge the walk starts from. + const edge = (direction === 'DESC' ? endDate : startDate).toISOString(); + const first = nextSearchWindow( + { ordinal: -1, startTime: edge, endTime: edge }, + { start: startDate, end: endDate }, + direction, + ); + return { window: first ?? wholeRange, windowed: true, cutPart, direction }; +} + +const WINDOW_SIZES_MS = DEFAULT_TIME_WINDOWS_SECONDS.map(sec => sec * 1000); +const LARGEST_WINDOW_MS = WINDOW_SIZES_MS[WINDOW_SIZES_MS.length - 1]; + +// Total length of the first `count` windows; the largest size repeats. +function cumulativeWindowMs(count: number): number { + const listed = WINDOW_SIZES_MS.slice(0, count).reduce((a, b) => a + b, 0); + return ( + listed + Math.max(0, count - WINDOW_SIZES_MS.length) * LARGEST_WINDOW_MS + ); } -// Floors boundaries between windows to the second, so a second-precision +// Boundaries between windows are floored to the second, so a second-precision // column such as TimestampTime doesn't drop rows at a boundary. The caller's -// bounds stay exact; windows left empty are dropped. -function snapBoundariesToSeconds( - windows: TimeWindow[], - startDate: Date, - endDate: Date, -): TimeWindow[] { - if (windows.length <= 1) return windows; +// bounds stay exact; a window that flooring empties is skipped. +const floorToSecond = (ms: number, startMs: number) => + Math.max(Math.floor(ms / 1000) * 1000, startMs); - const snap = (t: Date) => { - if ( - t.getTime() === startDate.getTime() || - t.getTime() === endDate.getTime() - ) { - return t; +/** + * The window after `current`, or null when the walk has reached the range + * edge. Its near edge is the current window's far edge, so windows never gap + * or overlap; its far edge is measured from the range edge, so flooring + * doesn't accumulate. + */ +export function nextSearchWindow( + current: CursorWindow, + range: Range, + direction: SearchDirection, +): CursorWindow | null { + const startMs = range.start.getTime(); + const endMs = range.end.getTime(); + let { ordinal } = current; + let ws = Date.parse(current.startTime); + let we = Date.parse(current.endTime); + for (;;) { + ordinal += 1; + if (direction === 'DESC') { + if (ws <= startMs) return null; + we = ws; + const raw = endMs - cumulativeWindowMs(ordinal + 1); + ws = raw <= startMs ? startMs : floorToSecond(raw, startMs); + } else { + if (we >= endMs) return null; + ws = we; + const raw = startMs + cumulativeWindowMs(ordinal + 1); + we = raw >= endMs ? endMs : floorToSecond(raw, startMs); } - return new Date( - Math.max(Math.floor(t.getTime() / 1000) * 1000, startDate.getTime()), - ); + if (ws < we) { + return { + ordinal, + startTime: new Date(ws).toISOString(), + endTime: new Date(we).toISOString(), + }; + } + } +} + +/** Windows are [start, end); the one holding the range end also includes it. */ +export function windowBounds( + window: CursorWindow, + range: Range, +): { + dateRange: [Date, Date]; + dateRangeStartInclusive: true; + dateRangeEndInclusive: boolean; +} { + const end = new Date(window.endTime); + return { + dateRange: [new Date(window.startTime), end], + dateRangeStartInclusive: true, + dateRangeEndInclusive: end.getTime() === range.end.getTime(), }; +} - return windows - .map(w => ({ - ...w, - startTime: snap(w.startTime), - endTime: snap(w.endTime), - })) - .filter(w => w.startTime.getTime() < w.endTime.getTime()) - .map((w, windowIndex) => ({ ...w, windowIndex })); +/** + * Shrinks a window to the rows not yet returned, cut at the last row's whole + * second, so the next page's offset only skips that second's returned rows. + * The far edge is untouched, so nextSearchWindow is unaffected. + */ +export function cutWindowAtSecond({ + window, + direction, + offset, + pageLength, + lastSecondMs, + trailingRunLength, +}: { + window: CursorWindow; + direction: SearchDirection; + offset: number; + pageLength: number; + lastSecondMs: number; + trailingRunLength: number; +}): { window: CursorWindow; offset: number } { + const startMs = Date.parse(window.startTime); + const endMs = Date.parse(window.endTime); + const cut = + direction === 'DESC' + ? { + ...window, + endTime: new Date(Math.min(lastSecondMs + 1000, endMs)).toISOString(), + } + : { + ...window, + startTime: new Date(Math.max(lastSecondMs, startMs)).toISOString(), + }; + // An unchanged window means the whole page was that one second. + const unchanged = + cut.startTime === window.startTime && cut.endTime === window.endTime; + return { + window: unchanged ? window : cut, + offset: unchanged ? offset + pageLength : trailingRunLength, + }; } diff --git a/packages/common-utils/src/__tests__/searchWindows.test.ts b/packages/common-utils/src/__tests__/searchWindows.test.ts deleted file mode 100644 index da5b35c064..0000000000 --- a/packages/common-utils/src/__tests__/searchWindows.test.ts +++ /dev/null @@ -1,75 +0,0 @@ -import { - generateTimeWindowsAscending, - generateTimeWindowsDescending, - windowInclusivity, -} from '@/core/searchWindows'; - -const START = new Date('2026-05-10T00:00:00.000Z'); -const END = new Date('2026-05-11T00:00:00.000Z'); - -describe('windowInclusivity', () => { - it('makes descending windows [start, end), newest window owning the range end', () => { - const windows = generateTimeWindowsDescending(START, END); - expect(windows.length).toBeGreaterThan(1); - - expect(windowInclusivity(windows[0], windows.length)).toEqual({ - dateRangeStartInclusive: true, - dateRangeEndInclusive: true, - }); - expect( - windowInclusivity(windows[windows.length - 1], windows.length), - ).toEqual({ - dateRangeStartInclusive: true, - dateRangeEndInclusive: false, - }); - }); - - it('makes ascending windows half-open, newest window owning the range end', () => { - const windows = generateTimeWindowsAscending(START, END); - expect(windows.length).toBeGreaterThan(1); - - expect(windowInclusivity(windows[0], windows.length)).toEqual({ - dateRangeStartInclusive: true, - dateRangeEndInclusive: false, - }); - expect( - windowInclusivity(windows[windows.length - 1], windows.length), - ).toEqual({ - dateRangeStartInclusive: true, - dateRangeEndInclusive: true, - }); - }); - - it('leaves a single window fully inclusive on both ends', () => { - const windows = generateTimeWindowsDescending(START, START); - expect(windows).toHaveLength(1); - expect(windowInclusivity(windows[0], 1)).toEqual({ - dateRangeStartInclusive: true, - dateRangeEndInclusive: true, - }); - }); - - it.each([ - ['descending', generateTimeWindowsDescending(START, END)], - ['ascending', generateTimeWindowsAscending(START, END)], - ])( - 'lets exactly one %s window claim each shared boundary instant', - (_, windows) => { - // Every window includes its start; only the one holding the range end - // includes its end. - const ownsRangeEnd = (w: (typeof windows)[number]) => - w.endTime.getTime() === END.getTime(); - const wrongClaimants = windows.filter( - w => - !ownsRangeEnd(w) && - windowInclusivity(w, windows.length).dateRangeEndInclusive, - ); - expect(wrongClaimants).toEqual([]); - expect( - windows.every( - w => windowInclusivity(w, windows.length).dateRangeStartInclusive, - ), - ).toBe(true); - }, - ); -}); diff --git a/packages/common-utils/src/__tests__/utils.test.ts b/packages/common-utils/src/__tests__/utils.test.ts index ee0ae6eae0..3eb16e0551 100644 --- a/packages/common-utils/src/__tests__/utils.test.ts +++ b/packages/common-utils/src/__tests__/utils.test.ts @@ -19,6 +19,7 @@ import { hasPositiveSeriesLimit, isFirstOrderByAscending, isJsonExpression, + isRoundedTimestampPart, isTimestampExpressionInFirstOrderBy, joinQuerySettings, optimizeTimestampValueExpression, @@ -2550,6 +2551,22 @@ describe('utils', () => { }); }); + describe('isRoundedTimestampPart', () => { + it.each([ + ['toStartOfMinute(Timestamp)', undefined, true], + ['toDate(Timestamp)', undefined, true], + ['EventDate', 'Date', true], + ['EventDate', 'LowCardinality(Date)', true], + ['EventDate', 'Nullable(Date32)', true], + ['TimestampTime', 'DateTime', false], + ['Timestamp', 'DateTime64(9)', false], + ['toDateTime(Timestamp)', undefined, false], + ['TODATE(Timestamp)', undefined, false], + ])('%s (%s) -> %s', (part, columnType, expected) => { + expect(isRoundedTimestampPart(part, columnType)).toBe(expected); + }); + }); + describe('parseToStartOfFunction', () => { it.each([ { diff --git a/packages/common-utils/src/core/renderChartConfig.ts b/packages/common-utils/src/core/renderChartConfig.ts index 8de7551337..0c3979984f 100644 --- a/packages/common-utils/src/core/renderChartConfig.ts +++ b/packages/common-utils/src/core/renderChartConfig.ts @@ -18,6 +18,7 @@ import { extractSettingsClauseFromEnd, getFirstTimestampValueExpression, hasPositiveSeriesLimit, + isRoundedTimestampPart, joinQuerySettings, optimizeTimestampValueExpression, parseToStartOfFunction, @@ -1065,15 +1066,12 @@ export async function timeFilterExpr({ const isDateType = /^Date(?:32)?$/i.test(columnType) || isToDateExpr; // toStartOf* and Date filters must stay inclusive — strict < on a rounded value drops a whole interval + const rounded = isRoundedTimestampPart(col, columnType); const startOp = - dateRangeStartInclusive || - toStartOf || - isDateType || - flooredPrecision != null + dateRangeStartInclusive || rounded || flooredPrecision != null ? '>=' : '>'; - const endOp = - dateRangeEndInclusive || toStartOf || isDateType ? '<=' : '<'; + const endOp = dateRangeEndInclusive || rounded ? '<=' : '<'; if (isDateType) { return chSql`(${unsafeTimestampValueExpression} ${startOp} toDate(${startTimeCond}) AND ${unsafeTimestampValueExpression} ${endOp} toDate(${endTimeCond}))`; diff --git a/packages/common-utils/src/core/searchChartConfig.ts b/packages/common-utils/src/core/searchChartConfig.ts index 37144db8ab..1e019ef697 100644 --- a/packages/common-utils/src/core/searchChartConfig.ts +++ b/packages/common-utils/src/core/searchChartConfig.ts @@ -91,7 +91,7 @@ export type SearchChartConfigInput = { * so a single non-empty check covers "skip empty `''` strings and skip empty * `[]` arrays" symmetrically — matching the app's historical behavior. */ -function resolveSelect( +export function resolveSelect( source: TSource, select: SelectList | null | undefined, ): BuilderChartConfig['select'] { diff --git a/packages/common-utils/src/core/searchWindows.ts b/packages/common-utils/src/core/searchWindows.ts index 57c6dce7cf..d7940d051e 100644 --- a/packages/common-utils/src/core/searchWindows.ts +++ b/packages/common-utils/src/core/searchWindows.ts @@ -101,28 +101,3 @@ export function generateTimeWindowsAscending( return windows; } - -/** - * Windows are [start, end), so a row on a shared boundary lands on one page. - * The window holding the caller's endTime also includes its end. - */ -export function windowInclusivity( - window: TimeWindow, - windowCount: number, -): { dateRangeStartInclusive: boolean; dateRangeEndInclusive: boolean } { - const isLast = window.windowIndex === windowCount - 1; - - if (window.direction === 'DESC') { - // Windows walk backwards: index 0 is newest, the last index is oldest. - return { - dateRangeStartInclusive: true, - dateRangeEndInclusive: window.windowIndex === 0, - }; - } - - // ASC: index 0 is oldest, the last index is newest. - return { - dateRangeStartInclusive: true, - dateRangeEndInclusive: isLast, - }; -} diff --git a/packages/common-utils/src/core/utils.ts b/packages/common-utils/src/core/utils.ts index 0042b9da81..bd28d49eaf 100644 --- a/packages/common-utils/src/core/utils.ts +++ b/packages/common-utils/src/core/utils.ts @@ -4,6 +4,8 @@ import { formatInTimeZone, zonedTimeToUtc } from 'date-fns-tz'; import { omit } from 'lodash'; import { z } from 'zod'; +import { stripTypeWrappers } from '@/core/eventDeltas'; + export { default as objectHash } from 'object-hash'; import { @@ -1164,6 +1166,21 @@ export const isFirstOrderByAscending = ( return !isDescending; }; +/** + * Whether timeFilterExpr filters this part of a timestamp expression with + * inclusive bounds on both ends: toStartOf*, toDate, or a Date/Date32 column. + */ +export function isRoundedTimestampPart( + part: string, + columnType: string | undefined, +): boolean { + return ( + parseToStartOfFunction(part) != null || + /^toDate\s*\(/.test(part) || + /^Date(?:32)?$/i.test(stripTypeWrappers(columnType ?? '')) + ); +} + /** * Parses a single expression of the form * `toStartOf(column[, timezone])` or `toStartOfInterval(column[, interval[, origin[, timezone]]])`. From a9e9025bf7d08908c969f2f5cf1daecb030d67c1 Mon Sep 17 00:00:00 2001 From: vinzee Date: Sun, 25 Oct 2026 19:51:55 -0700 Subject: [PATCH 3/3] feat(api): window cursor walks led by a minute-rounded sort key Walks sorted by toStartOfSecond/Minute/FiveMinutes/FifteenMinutes, with an exact timestamp part in the source expression, now use windows floored to the key's bucket and cut at the last row's bucket. These buckets divide 15 minutes, so they align to UTC in every current time zone. Other rounded keys keep a single window. --- .../external-api/__tests__/search.int.test.ts | 70 +++++++++++++ .../utils/__tests__/searchWindowCut.test.ts | 41 ++++++-- .../v2/utils/__tests__/searchWindows.test.ts | 86 ++++++++++++---- .../routers/external-api/v2/utils/search.ts | 38 ++++--- .../external-api/v2/utils/searchWindows.ts | 99 +++++++++++++------ 5 files changed, 261 insertions(+), 73 deletions(-) diff --git a/packages/api/src/routers/external-api/__tests__/search.int.test.ts b/packages/api/src/routers/external-api/__tests__/search.int.test.ts index 3e7c836326..f9f349c121 100644 --- a/packages/api/src/routers/external-api/__tests__/search.int.test.ts +++ b/packages/api/src/routers/external-api/__tests__/search.int.test.ts @@ -984,6 +984,76 @@ describe('External API v2 Search', () => { }, ); + it('pages a minute-rounded sort in order with minute-aligned windows and cuts', async () => { + const minuteSource = await Source.create({ + kind: SourceKind.Log, + team: logSource.team, + from: logSource.from, + defaultTableSelectExpression: 'Timestamp, SeverityText, Body', + timestampValueExpression: 'toStartOfMinute(Timestamp), Timestamp', + connection: logSource.connection, + name: 'Logs (minute key)', + }); + // 20 rows a minute for 30 minutes, so the walk crosses the first window's + // boundary; Body order is unrelated to time within each minute. + const base = DEFAULT_END_TIME - 30 * 60 * 1000; + await bulkInsertLogs( + Array.from({ length: 600 }, (_, i) => ({ + ServiceName: 'cursor-minute-test', + Timestamp: new Date(base + i * 3000), + SeverityText: 'INFO', + Body: `row-${String((i * 37) % 600).padStart(3, '0')}`, + })), + ); + + const rows: { minute: number; body: string }[] = []; + const cursors: { + ws: string; + we: string; + o: number; + s: string; + e: string; + }[] = []; + let cursor: string | null | undefined; + for (let pages = 0; pages < 100; pages++) { + const res = await search({ + ...cursorBody(), + sourceId: minuteSource.id.toString(), + where: 'ServiceName:"cursor-minute-test"', + orderBy: 'toStartOfMinute(Timestamp) DESC, Body ASC', + maxResults: 50, + cursor: cursor ?? 'start', + }).expect(200); + for (const r of res.body.data) { + const minute = Math.floor(Date.parse(r.Timestamp) / 60_000); + rows.push({ minute, body: r.Body }); + } + cursor = res.body.nextCursor; + if (cursor == null) break; + cursors.push( + JSON.parse(Buffer.from(cursor.split('.')[0], 'base64url').toString()), + ); + } + + expect(cursor).toBeNull(); + expect(new Set(rows.map(r => r.body)).size).toBe(600); + rows.slice(1).forEach((r, i) => { + const prev = rows.at(i); + expect(prev).toBeDefined(); + if (prev && r.minute === prev.minute) + expect(r.body > prev.body).toBe(true); + else if (prev) expect(r.minute < prev.minute).toBe(true); + }); + for (const c of cursors) { + expect(c.o).toBeLessThanOrEqual(20); + for (const bound of [c.ws, c.we]) { + if (bound !== c.s && bound !== c.e) { + expect(Date.parse(bound) % 60_000).toBe(0); + } + } + } + }); + it('rejects a select that uses the reserved cursor column name', async () => { await search({ ...cursorBody(), diff --git a/packages/api/src/routers/external-api/v2/utils/__tests__/searchWindowCut.test.ts b/packages/api/src/routers/external-api/v2/utils/__tests__/searchWindowCut.test.ts index 44f3311a84..18d1ffc2e4 100644 --- a/packages/api/src/routers/external-api/v2/utils/__tests__/searchWindowCut.test.ts +++ b/packages/api/src/routers/external-api/v2/utils/__tests__/searchWindowCut.test.ts @@ -1,4 +1,4 @@ -import { cutWindowAtSecond } from '@/routers/external-api/v2/utils/searchWindows'; +import { cutWindowAtBucket } from '@/routers/external-api/v2/utils/searchWindows'; const window = { ordinal: 0, @@ -7,15 +7,16 @@ const window = { }; const ms = (iso: string) => Date.parse(iso); -describe('cutWindowAtSecond', () => { +describe('cutWindowAtBucket', () => { it('ends a descending window just after the last row second', () => { expect( - cutWindowAtSecond({ + cutWindowAtBucket({ window, direction: 'DESC', offset: 0, pageLength: 2000, - lastSecondMs: ms('2026-10-25T11:58:31.000Z'), + lastBucketMs: ms('2026-10-25T11:58:31.000Z'), + bucketMs: 1000, trailingRunLength: 12, }), ).toEqual({ @@ -26,12 +27,13 @@ describe('cutWindowAtSecond', () => { it('keeps the range end when the last row second reaches it', () => { expect( - cutWindowAtSecond({ + cutWindowAtBucket({ window, direction: 'DESC', offset: 0, pageLength: 2000, - lastSecondMs: ms('2026-10-25T12:00:00.000Z'), + lastBucketMs: ms('2026-10-25T12:00:00.000Z'), + bucketMs: 1000, trailingRunLength: 2000, }), ).toEqual({ window, offset: 2000 }); @@ -39,12 +41,13 @@ describe('cutWindowAtSecond', () => { it('starts an ascending window at the last row second', () => { expect( - cutWindowAtSecond({ + cutWindowAtBucket({ window, direction: 'ASC', offset: 0, pageLength: 2000, - lastSecondMs: ms('2026-10-25T11:46:10.000Z'), + lastBucketMs: ms('2026-10-25T11:46:10.000Z'), + bucketMs: 1000, trailingRunLength: 7, }), ).toEqual({ @@ -53,6 +56,23 @@ describe('cutWindowAtSecond', () => { }); }); + it.each([ + ['DESC', { ...window, endTime: '2026-10-25T11:59:00.000Z' }], + ['ASC', { ...window, startTime: '2026-10-25T11:58:00.000Z' }], + ] as const)('cuts a %s window at a minute bucket', (direction, expected) => { + expect( + cutWindowAtBucket({ + window, + direction, + offset: 0, + pageLength: 2000, + lastBucketMs: ms('2026-10-25T11:58:00.000Z'), + bucketMs: 60_000, + trailingRunLength: 40, + }), + ).toEqual({ window: expected, offset: 40 }); + }); + it.each(['DESC', 'ASC'] as const)( 'adds the page to the offset when the whole %s page was one second', direction => { @@ -61,12 +81,13 @@ describe('cutWindowAtSecond', () => { ? { ...window, endTime: '2026-10-25T11:58:32.000Z' } : { ...window, startTime: '2026-10-25T11:58:31.000Z' }; expect( - cutWindowAtSecond({ + cutWindowAtBucket({ window: cut, direction, offset: 300, pageLength: 1000, - lastSecondMs: ms('2026-10-25T11:58:31.000Z'), + lastBucketMs: ms('2026-10-25T11:58:31.000Z'), + bucketMs: 1000, trailingRunLength: 1000, }), ).toEqual({ window: cut, offset: 1300 }); diff --git a/packages/api/src/routers/external-api/v2/utils/__tests__/searchWindows.test.ts b/packages/api/src/routers/external-api/v2/utils/__tests__/searchWindows.test.ts index 7041738313..a31cb846ed 100644 --- a/packages/api/src/routers/external-api/v2/utils/__tests__/searchWindows.test.ts +++ b/packages/api/src/routers/external-api/v2/utils/__tests__/searchWindows.test.ts @@ -37,7 +37,7 @@ async function resolveSearchWindows( for ( let w: typeof first.window | null = first.window; w != null; - w = nextSearchWindow(w, range, first.direction) + w = nextSearchWindow(w, range, first.direction, first.cut?.bucketMs) ) { windows.push({ startTime: new Date(w.startTime), @@ -51,7 +51,12 @@ async function resolveSearchWindows( // Reference windows: build every window, floor the boundaries between // them to the second, drop empties. Stepping must reproduce it exactly. -function referenceWindows(start: Date, end: Date, direction: 'ASC' | 'DESC') { +function referenceWindows( + start: Date, + end: Date, + direction: 'ASC' | 'DESC', + alignMs = 1000, +) { const windows = direction === 'ASC' ? generateTimeWindowsAscending(start, end) @@ -60,7 +65,7 @@ function referenceWindows(start: Date, end: Date, direction: 'ASC' | 'DESC') { const snap = (t: Date) => t.getTime() === start.getTime() || t.getTime() === end.getTime() ? t.getTime() - : Math.max(Math.floor(t.getTime() / 1000) * 1000, start.getTime()); + : Math.max(Math.floor(t.getTime() / alignMs) * alignMs, start.getTime()); return windows .map(w => [snap(w.startTime), snap(w.endTime)]) .filter(([a, b]) => a < b) @@ -225,9 +230,25 @@ describe('search window stepping', () => { }, ); - it.each(['DESC', 'ASC'] as const)( - 'steps through exactly the reference windows (%s)', - async direction => { + it.each([ + ['DESC', 'Timestamp', 'Timestamp', 1000], + ['ASC', 'Timestamp', 'Timestamp', 1000], + [ + 'DESC', + 'toStartOfMinute(Timestamp), Timestamp', + 'toStartOfMinute(Timestamp)', + 60_000, + ], + [ + 'ASC', + 'toStartOfMinute(Timestamp), Timestamp', + 'toStartOfMinute(Timestamp)', + 60_000, + ], + ] as const)( + 'steps through exactly the reference windows (%s, %s)', + async (direction, timestampValueExpression, sortKey, alignMs) => { + const walkSource = { timestampValueExpression }; // Deterministic pseudo-random ranges: 0 to 30 days, millisecond precision. let seed = 42; const rand = () => @@ -238,16 +259,16 @@ describe('search window stepping', () => { let cumulative = 0; for (const size of [...DEFAULT_TIME_WINDOWS_SECONDS, 86_400, 86_400]) { cumulative += size * 1000; - for (const delta of [-1300, -300, 0, 300, 1300]) { + for (const delta of [-61_300, -1300, -300, 0, 300, 1300, 61_300]) { const end = new Date(edgeStart.getTime() + cumulative + delta); const walked = await resolveSearchWindows( - source, - `Timestamp ${direction}`, + walkSource, + `${sortKey} ${direction}`, edgeStart, end, ); expect(walked.map(w => [w.startTime, w.endTime])).toEqual( - referenceWindows(edgeStart, end, direction), + referenceWindows(edgeStart, end, direction, alignMs), ); } } @@ -259,13 +280,13 @@ describe('search window stepping', () => { start.getTime() + Math.floor(rand() * 30 * 86_400_000), ); const walked = await resolveSearchWindows( - source, - `Timestamp ${direction}`, + walkSource, + `${sortKey} ${direction}`, start, end, ); expect(walked.map(w => [w.startTime, w.endTime])).toEqual( - referenceWindows(start, end, direction), + referenceWindows(start, end, direction, alignMs), ); } }, @@ -318,20 +339,47 @@ describe('search window stepping', () => { }, ); expect(first.windowed).toBe(true); - expect(first.cutPart).toBe(cutPart); + expect(first.cut).toEqual({ expression: cutPart, bucketMs: 1000 }); }, ); it.each([ + ['toStartOfSecond(Timestamp)', 1000], + ['toStartOfMinute(Timestamp)', 60_000], + ['toStartOfFiveMinutes(Timestamp)', 300_000], + ['toStartOfFiveMinute(Timestamp)', 300_000], + ['toStartOfFifteenMinutes(Timestamp)', 900_000], + ])( + 'cuts on a rounded leading key %s in %ims buckets', + async (key, bucketMs) => { + const first = await resolveFirstSearchWindow( + { timestampValueExpression: `${key}, Timestamp` }, + `${key} DESC, Body`, + START, + END, + { columnType: async () => 'DateTime64(9)' }, + ); + expect(first.windowed).toBe(true); + expect(first.cut).toEqual({ expression: key, bucketMs }); + }, + ); + + it.each([ + // Ten-minute buckets don't align to UTC in +x:45 time zones. [ - 'toStartOfMinute(Timestamp), Timestamp', - 'toStartOfMinute(Timestamp) DESC, Body', + 'toStartOfTenMinutes(Timestamp), Timestamp', + 'toStartOfTenMinutes(Timestamp) DESC, Body', + ], + [ + 'toStartOfHour(Timestamp), Timestamp', + 'toStartOfHour(Timestamp) DESC, Body', ], ['EventDate, EventTime', 'EventDate DESC, Body'], + // Without an exact part, rounded bounds are inclusive on both ends. + ['toStartOfMinute(Timestamp)', 'toStartOfMinute(Timestamp) DESC, Body'], ])( - 'uses one window for %s ordered by a rounded key (%s)', + 'uses one window for %s ordered by %s', async (timestampValueExpression, orderBy) => { - // A window boundary inside one rounded key would break the requested order. const first = await resolveFirstSearchWindow( { timestampValueExpression }, orderBy, @@ -343,7 +391,7 @@ describe('search window stepping', () => { }, ); expect(first.windowed).toBe(false); - expect(first.cutPart).toBeNull(); + expect(first.cut).toBeNull(); }, ); diff --git a/packages/api/src/routers/external-api/v2/utils/search.ts b/packages/api/src/routers/external-api/v2/utils/search.ts index e0c5215f02..a744412462 100644 --- a/packages/api/src/routers/external-api/v2/utils/search.ts +++ b/packages/api/src/routers/external-api/v2/utils/search.ts @@ -30,7 +30,7 @@ import { type SearchWalk, } from './searchCursor'; import { - cutWindowAtSecond, + cutWindowAtBucket, nextSearchWindow, resolveFirstSearchWindow, windowBounds, @@ -225,12 +225,12 @@ export async function runSearchConfig({ const range = { start: walkStart, end: walkEnd }; const window = cursorState?.window ?? first.window; const effectiveOffset = cursorState ? cursorState.offset : offset; - const cutColumn = cursorState ? first.cutPart : null; + const cut = cursorState ? first.cut : null; // A walk that can't cut its windows pages with plain offsets, so it keeps // the `offset` field's cap. Erroring, rather than ending the walk, keeps a // truncated result from looking complete. const requestedLimit = - cursorState && !cutColumn + cursorState && !cut ? Math.min(maxResults, MAX_SINGLE_WINDOW_OFFSET - effectiveOffset) : maxResults; // A cursor walk reads one extra row to learn whether the window continues, @@ -255,8 +255,8 @@ export async function runSearchConfig({ const searchBase = buildSearchChartConfig(source, { where: typeof config.where === 'string' ? config.where : '', whereLanguage: config.whereLanguage ?? 'lucene', - select: cutColumn - ? `${baseSelect || '*'}, toUnixTimestamp64Milli(toDateTime64(${cutColumn}, 3)) AS ${CURSOR_TS_COLUMN}` + select: cut + ? `${baseSelect || '*'}, toUnixTimestamp64Milli(toDateTime64(${cut.expression}, 3)) AS ${CURSOR_TS_COLUMN}` : (config.select ?? null), displayType: DisplayType.Search, orderBy: effectiveOrderBy, @@ -295,16 +295,16 @@ export async function runSearchConfig({ (rows.length > pageLimit || (queryLimit === pageLimit && rows.length === pageLimit)); const page = cursorState ? rows.slice(0, pageLimit) : rows; - const split = cutColumn + const split = cut ? page.map(({ [CURSOR_TS_COLUMN]: ms, ...row }) => { - const second = Math.floor(Number(ms) / 1000) * 1000; - if (!Number.isFinite(second)) { + const bucket = Math.floor(Number(ms) / cut.bucketMs) * cut.bucketMs; + if (!Number.isFinite(bucket)) { throw new Error(`Search result is missing ${CURSOR_TS_COLUMN}`); } - return { second, row }; + return { bucket, row }; }) : null; - const seconds = split?.map(r => r.second) ?? []; + const buckets = split?.map(r => r.bucket) ?? []; const data = split?.map(r => r.row) ?? page; // Only a cursor walk gets a next cursor; offset-only callers are unchanged. @@ -314,19 +314,20 @@ export async function runSearchConfig({ startTime: cursorState.startTime, endTime: cursorState.endTime, }; - if (hasMoreInWindow && cutColumn) { - const lastSecond = seconds[seconds.length - 1]; + if (hasMoreInWindow && cut) { + const lastBucket = buckets[buckets.length - 1]; let trailingRunLength = 0; - while (seconds[seconds.length - 1 - trailingRunLength] === lastSecond) { + while (buckets[buckets.length - 1 - trailingRunLength] === lastBucket) { trailingRunLength++; } nextCursorState = { - ...cutWindowAtSecond({ + ...cutWindowAtBucket({ window, direction: first.direction, offset: effectiveOffset, pageLength: data.length, - lastSecondMs: lastSecond, + lastBucketMs: lastBucket, + bucketMs: cut.bucketMs, trailingRunLength, }), ...pinned, @@ -338,7 +339,12 @@ export async function runSearchConfig({ ...pinned, }; } else { - const nextWindow = nextSearchWindow(window, range, first.direction); + const nextWindow = nextSearchWindow( + window, + range, + first.direction, + first.cut?.bucketMs, + ); if (nextWindow) { nextCursorState = { window: nextWindow, offset: 0, ...pinned }; } diff --git a/packages/api/src/routers/external-api/v2/utils/searchWindows.ts b/packages/api/src/routers/external-api/v2/utils/searchWindows.ts index 95b9a789c1..6157a4eb2a 100644 --- a/packages/api/src/routers/external-api/v2/utils/searchWindows.ts +++ b/packages/api/src/routers/external-api/v2/utils/searchWindows.ts @@ -3,6 +3,7 @@ import { getFirstOrderingItem, isFirstOrderByAscending, isRoundedTimestampPart, + parseToStartOfFunction, removeTrailingDirection, splitAndTrimWithBracket, } from '@hyperdx/common-utils/dist/core/utils'; @@ -42,6 +43,41 @@ async function isExactTimestampPart( export type SearchDirection = 'ASC' | 'DESC'; type Range = { start: Date; end: Date }; +/** The leading sort key windows are cut on, and the length of its buckets. */ +export type SearchCut = { expression: string; bucketMs: number }; + +// Rounded leading keys whose buckets divide 15 minutes. Every current UTC +// offset is a multiple of 15 minutes, so these buckets align to UTC in any +// time zone and windows can be floored to the bucket length. +const ROUNDED_KEY_BUCKET_MS: Record = { + toStartOfSecond: 1000, + toStartOfMinute: 60_000, + toStartOfFiveMinute: 300_000, + toStartOfFiveMinutes: 300_000, + toStartOfFifteenMinutes: 900_000, +}; + +async function resolveCut( + leading: string, + timestampValueExpression: string, + columnType: (column: string) => Promise, +): Promise { + if (await isExactTimestampPart(leading, columnType)) { + return { expression: leading, bucketMs: 1000 }; + } + const bucketMs = + ROUNDED_KEY_BUCKET_MS[parseToStartOfFunction(leading)?.function ?? '']; + if (bucketMs == null) return null; + // Rounded bounds are inclusive on both ends; an exact part's strict bound + // is what keeps windows half-open. + for (const part of splitAndTrimWithBracket(timestampValueExpression)) { + if (await isExactTimestampPart(part, columnType)) { + return { expression: leading, bucketMs }; + } + } + return null; +} + export async function resolveFirstSearchWindow( source: { timestampValueExpression?: string }, orderBy: string, @@ -57,8 +93,7 @@ export async function resolveFirstSearchWindow( ): Promise<{ window: CursorWindow; windowed: boolean; - /** The exact timestamp part leading the sort; windows are cut on it. */ - cutPart: string | null; + cut: SearchCut | null; direction: SearchDirection; }> { const direction = isFirstOrderByAscending(orderBy) ? 'ASC' : 'DESC'; @@ -71,17 +106,18 @@ export async function resolveFirstSearchWindow( orderBy, source.timestampValueExpression, ); - // Windows need an exact leading key: a boundary inside one rounded key - // (a minute, a day) would split rows the sort keeps together. - const cutPart = - !forceSingleWindow && - startDate < endDate && - leading != null && - (await isExactTimestampPart(leading, columnType)) - ? leading + // Window boundaries are floored to the leading key's bucket, so one bucket's + // rows, which the sort keeps together, never span two windows. + const cut = + !forceSingleWindow && startDate < endDate && leading != null + ? await resolveCut( + leading, + source.timestampValueExpression ?? '', + columnType, + ) : null; - if (cutPart == null) { - return { window: wholeRange, windowed: false, cutPart, direction }; + if (cut == null) { + return { window: wholeRange, windowed: false, cut, direction }; } // Step from an empty window at the edge the walk starts from. @@ -90,8 +126,9 @@ export async function resolveFirstSearchWindow( { ordinal: -1, startTime: edge, endTime: edge }, { start: startDate, end: endDate }, direction, + cut.bucketMs, ); - return { window: first ?? wholeRange, windowed: true, cutPart, direction }; + return { window: first ?? wholeRange, windowed: true, cut, direction }; } const WINDOW_SIZES_MS = DEFAULT_TIME_WINDOWS_SECONDS.map(sec => sec * 1000); @@ -105,11 +142,12 @@ function cumulativeWindowMs(count: number): number { ); } -// Boundaries between windows are floored to the second, so a second-precision -// column such as TimestampTime doesn't drop rows at a boundary. The caller's -// bounds stay exact; a window that flooring empties is skipped. -const floorToSecond = (ms: number, startMs: number) => - Math.max(Math.floor(ms / 1000) * 1000, startMs); +// Boundaries between windows are floored to the bucket (a second for an exact +// key), so a second-precision column such as TimestampTime doesn't drop rows +// at a boundary. The caller's bounds stay exact; a window that flooring +// empties is skipped. +const floorToBucket = (ms: number, startMs: number, bucketMs: number) => + Math.max(Math.floor(ms / bucketMs) * bucketMs, startMs); /** * The window after `current`, or null when the walk has reached the range @@ -121,6 +159,7 @@ export function nextSearchWindow( current: CursorWindow, range: Range, direction: SearchDirection, + bucketMs = 1000, ): CursorWindow | null { const startMs = range.start.getTime(); const endMs = range.end.getTime(); @@ -133,12 +172,12 @@ export function nextSearchWindow( if (ws <= startMs) return null; we = ws; const raw = endMs - cumulativeWindowMs(ordinal + 1); - ws = raw <= startMs ? startMs : floorToSecond(raw, startMs); + ws = raw <= startMs ? startMs : floorToBucket(raw, startMs, bucketMs); } else { if (we >= endMs) return null; ws = we; const raw = startMs + cumulativeWindowMs(ordinal + 1); - we = raw >= endMs ? endMs : floorToSecond(raw, startMs); + we = raw >= endMs ? endMs : floorToBucket(raw, startMs, bucketMs); } if (ws < we) { return { @@ -168,23 +207,25 @@ export function windowBounds( } /** - * Shrinks a window to the rows not yet returned, cut at the last row's whole - * second, so the next page's offset only skips that second's returned rows. - * The far edge is untouched, so nextSearchWindow is unaffected. + * Shrinks a window to the rows not yet returned, cut at the last row's bucket, + * so the next page's offset only skips that bucket's returned rows. The far + * edge is untouched, so nextSearchWindow is unaffected. */ -export function cutWindowAtSecond({ +export function cutWindowAtBucket({ window, direction, offset, pageLength, - lastSecondMs, + lastBucketMs, + bucketMs, trailingRunLength, }: { window: CursorWindow; direction: SearchDirection; offset: number; pageLength: number; - lastSecondMs: number; + lastBucketMs: number; + bucketMs: number; trailingRunLength: number; }): { window: CursorWindow; offset: number } { const startMs = Date.parse(window.startTime); @@ -193,11 +234,13 @@ export function cutWindowAtSecond({ direction === 'DESC' ? { ...window, - endTime: new Date(Math.min(lastSecondMs + 1000, endMs)).toISOString(), + endTime: new Date( + Math.min(lastBucketMs + bucketMs, endMs), + ).toISOString(), } : { ...window, - startTime: new Date(Math.max(lastSecondMs, startMs)).toISOString(), + startTime: new Date(Math.max(lastBucketMs, startMs)).toISOString(), }; // An unchanged window means the whole page was that one second. const unchanged =