diff --git a/packages/mosaic/core/docs/latest-only-streams.md b/packages/mosaic/core/docs/latest-only-streams.md new file mode 100644 index 000000000..df74623c0 --- /dev/null +++ b/packages/mosaic/core/docs/latest-only-streams.md @@ -0,0 +1,24 @@ +# Latest-Only Stream Scheduling + +## Problem +Interactive applications often generate high-frequency query streams (e.g., brushing, slider dragging). Sending every intermediate query to the server can cause: +1. **Backlog Latency:** The server queues up stale requests, delaying the "latest" relevant result. +2. **Wasted Work:** Computing results for chart states that the user has already passed. + +## API +The `Coordinator.query()` method accepts two new options: + +```typescript +coord.query(sql, { + stream: 'my-interaction-id', // Group requests by this ID + latest: true // Enable pruning of stale requests +}); +``` + +## Semantics +When `latest: true` is set: +1. **Queue Pruning:** Any *pending* (queued but not yet sent) requests with the same `stream` ID are removed from the client-side queue. +2. **Stale Suppression:** If a request was already sent (inflight) and returns *after* a newer request was issued, its result is rejected with `"Stale"` to prevent UI jitter. + +## Limitations +* **No Server Cancellation:** This feature does not interrupt queries already executing on the database. It only prevents new stale queries from being sent and suppresses stale results. diff --git a/packages/mosaic/core/src/Coordinator.ts b/packages/mosaic/core/src/Coordinator.ts index 104c65b39..62d9f8538 100644 --- a/packages/mosaic/core/src/Coordinator.ts +++ b/packages/mosaic/core/src/Coordinator.ts @@ -166,6 +166,8 @@ export class Coordinator { * @param options.cache If true, cache the query result client-side within the QueryManager. * @param options.persist If true, request the database server to persist a cached query server-side. * @param options.priority The query priority, defaults to `Priority.Normal`. + * @param options.stream Stream ID for grouping related requests (e.g., 'brush'). + * @param options.latest If true, keeps only the most recent pending request for the stream ID. Older queued requests are pruned. * @returns A query result promise. */ query( @@ -175,6 +177,8 @@ export class Coordinator { cache?: boolean; persist?: boolean; priority?: number; + stream?: string; + latest?: boolean; [key: string]: unknown; } ): QueryResult; @@ -185,6 +189,8 @@ export class Coordinator { cache?: boolean; persist?: boolean; priority?: number; + stream?: string; + latest?: boolean; [key: string]: unknown; } ): QueryResult; @@ -195,6 +201,8 @@ export class Coordinator { cache?: boolean; persist?: boolean; priority?: number; + stream?: string; + latest?: boolean; [key: string]: unknown; } = {} ): QueryResult { @@ -202,9 +210,11 @@ export class Coordinator { type = 'arrow', cache = true, priority = Priority.Normal, + stream, + latest, ...otherOptions } = options; - return this.manager.request({ type, query, cache, options: otherOptions }, priority); + return this.manager.request({ type, query, cache, stream, latest, options: otherOptions }, priority); } /** @@ -217,11 +227,11 @@ export class Coordinator { */ prefetch( query: QueryType, - options?: { type?: 'arrow'; [key: string]: unknown } + options?: { type?: 'arrow';[key: string]: unknown } ): QueryResult
prefetch( query: QueryType, - options?: { type?: 'json'; [key: string]: unknown } + options?: { type?: 'json';[key: string]: unknown } ): QueryResult prefetch( query: QueryType, diff --git a/packages/mosaic/core/src/QueryManager.ts b/packages/mosaic/core/src/QueryManager.ts index 450956211..b89651581 100644 --- a/packages/mosaic/core/src/QueryManager.ts +++ b/packages/mosaic/core/src/QueryManager.ts @@ -19,6 +19,7 @@ export class QueryManager { public pendingResults: QueryResult[]; private maxConcurrentRequests: number; private pendingExec: boolean; + private streamGen = new Map(); constructor(maxConcurrentRequests: number = 32) { this.queue = new PriorityQueue(3); @@ -103,6 +104,14 @@ export class QueryManager { const data = await promise; + if (request.stream && request.latest) { + if (request.streamGen !== this.streamGen.get(request.stream)) { + this._logger.debug('Suppressing stale result', { stream: request.stream, gen: request.streamGen }); + result.reject('Stale'); + return; + } + } + if (cache) this.clientCache!.set(sql!, data); this._logger.debug(`Request: ${(performance.now() - t0).toFixed(1)}`); @@ -179,6 +188,27 @@ export class QueryManager { request(request: QueryRequest, priority: number = Priority.Normal): QueryResult { const result = new QueryResult(); const entry = { request, result }; + + if (request.stream && request.latest) { + // Latest-Only Scheduling: + // 1. Increment generation counter for this stream. + // 2. Prune any older queued requests for the same stream. + // 3. Mark request with generationID so stale results can be suppressed. + // Note: Inflight requests on the server are NOT cancelled. + const gen = (this.streamGen.get(request.stream) ?? 0) + 1; + this.streamGen.set(request.stream, gen); + request.streamGen = gen; + + this.queue.remove(entry => { + const req = entry.request; + if (req.stream === request.stream && req.latest) { + entry.result.reject('Stale'); + return true; + } + return false; + }); + } + if (this._consolidate) { this._consolidate.add(entry, priority); } else { diff --git a/packages/mosaic/core/src/types.ts b/packages/mosaic/core/src/types.ts index 028808e3c..73bbde908 100644 --- a/packages/mosaic/core/src/types.ts +++ b/packages/mosaic/core/src/types.ts @@ -7,6 +7,9 @@ export interface QueryRequest { query: string | Query | DescribeQuery; cache?: boolean; options?: Record; + stream?: string; + latest?: boolean; + streamGen?: number; } /** Type for an entry within a query manager. */ diff --git a/packages/mosaic/core/test/querymanager-latest-only.test.ts b/packages/mosaic/core/test/querymanager-latest-only.test.ts new file mode 100644 index 000000000..3b233fc52 --- /dev/null +++ b/packages/mosaic/core/test/querymanager-latest-only.test.ts @@ -0,0 +1,66 @@ +import { describe, it, expect } from 'vitest'; +import { QueryManager, Priority } from '../src/QueryManager.js'; +import type { Connector } from '../src/connectors/Connector.js'; + +// Mock connector that counts how many times query() is actually invoked. +// Adds a small delay so queued requests build up and pruning can occur. +class CountingConnector implements Connector { + public calls: number = 0; + + async query(req: any): Promise { + this.calls += 1; + // Delay to ensure queue backlog exists (so pruning matters) + await new Promise(r => setTimeout(r, 50)); + + // Return something valid for json requests + if (req.type === 'json') return [{ ok: true }]; + if (req.type === 'exec') return; + // Arrow is not needed here; if used, return empty + return []; + } +} + +describe('QueryManager latest-only scheduling', () => { + it('prunes stale requests and executes only a small number of connector calls', async () => { + const connector = new CountingConnector(); + + // Concurrency = 1 to force a backlog + const qm = new QueryManager(1); + qm.connector(connector); + + // Disable consolidation/caching to isolate latest-only behavior + qm.cache(false); + qm.consolidate(false); + + const N = 30; + + // Fire N requests quickly: same stream, latest-only. + // The correct implementation should reject stale ones and only fulfill the latest. + const reqs = Array.from({ length: N }, (_, i) => { + return qm.request( + { + type: 'json', + query: `SELECT ${i} AS qid`, + cache: false, + stream: 'ui', + latest: true + // streamGen: (only add if your implementation requires it; Step 13.1 will tell us) + }, + Priority.Normal + ); + }); + + const settled = await Promise.allSettled(reqs as any); + + const fulfilled = settled.filter(x => x.status === 'fulfilled').length; + const rejected = settled.filter(x => x.status === 'rejected').length; + + // Core correctness: only the newest should matter. + expect(fulfilled).toBe(1); + expect(rejected).toBe(N - 1); + + // Efficiency: we must not hit the connector N times. + // Be strict but realistic: should be single-digit calls, not 30. + expect(connector.calls).toBeLessThan(10); + }); +});