Line data Source code
1 : import * as flatbuffers from 'flatbuffers';
2 : import * as buffers from '@ankoh/dashql-core/buffers';
3 : import type { DashQL, DashQLCatalog, DashQLScript } from '@ankoh/dashql-core';
4 : import type { BatchRequest, BatchResult } from '@ankoh/dashql-core';
5 : import { createBatchWorker, type DashQLBatchWorker } from '@ankoh/dashql-core/browser-worker';
6 :
7 : export type BatchProcessor = (request: BatchRequest) => Promise<BatchResult>;
8 :
9 : /// SQL of the connection catalog, published as a descriptor in the connection layer. A script keeps the text, an entry
10 : /// id only gets the descriptor.
11 : export type CatalogBatchEntry = ({ script: DashQLScript; entryId?: undefined } | { entryId: number; script?: undefined }) & {
12 : text: string;
13 : requireFunctions?: boolean;
14 : };
15 :
16 1 : function entryIdOf(entry: CatalogBatchEntry): number {
17 1 : return entry.script?.catalog_entry_id ?? entry.entryId!;
18 1 : }
19 :
20 : export interface CatalogBatchOptions {
21 : abortSignal?: AbortSignal;
22 : isCurrent?: () => boolean;
23 : processor?: BatchProcessor;
24 : }
25 :
26 1 : const workers = new WeakMap<DashQL, DashQLBatchWorker>();
27 1 : const lifetimes = new WeakMap<DashQL, AbortController>();
28 1 : const generations = new WeakMap<DashQLCatalog, number>();
29 :
30 1 : function getCatalogBatchWorker(core: DashQL): DashQLBatchWorker {
31 1 : let worker = workers.get(core);
32 1 : if (worker == null || (!worker.isInitialized && !worker.isClosed)) {
33 1 : throw new Error('Core batch worker is not initialized; enable setupBatchWorker during initial setup');
34 : }
35 : // A worker closes after a failure, e.g. a panic of its core, the next batch runs on a new one
36 1 : if (worker.isClosed) {
37 1 : worker = createBatchWorker();
38 1 : workers.set(core, worker);
39 1 : }
40 1 : return worker;
41 1 : }
42 :
43 1 : export async function initializeCatalogBatchWorker(core: DashQL): Promise<void> {
44 1 : lifetimes.get(core)?.signal.throwIfAborted();
45 1 : let worker = workers.get(core);
46 1 : if (!worker) {
47 1 : worker = createBatchWorker();
48 1 : workers.set(core, worker);
49 1 : lifetimes.set(core, new AbortController());
50 1 : }
51 1 : await worker.initialize();
52 1 : }
53 :
54 : /// The core provider explicitly sets up and owns the worker lifecycle.
55 1 : export function disposeCatalogBatchWorker(core: DashQL): void {
56 : // Keep the aborted signal for response continuations already queued before disposal.
57 1 : lifetimes.get(core)?.abort();
58 1 : workers.get(core)?.dispose();
59 1 : workers.delete(core);
60 1 : }
61 :
62 1 : export function invalidateCatalogBatch(catalog: DashQLCatalog): void {
63 1 : generations.set(catalog, (generations.get(catalog) ?? 0) + 1);
64 1 : }
65 :
66 : /// Begin before fetching metadata, not after it, so a slow fetch cannot supersede a newer refresh.
67 1 : export function beginCatalogBatch(catalog: DashQLCatalog, options: CatalogBatchOptions = {}): () => void {
68 1 : invalidateCatalogBatch(catalog);
69 1 : const generation = generations.get(catalog);
70 1 : return () => {
71 1 : if (options.abortSignal?.aborted || generations.get(catalog) !== generation || options.isCurrent?.() === false) {
72 1 : throw new DOMException('Catalog refresh was cancelled or superseded', 'AbortError');
73 : }
74 1 : };
75 1 : }
76 :
77 1 : function batchRequest(entries: CatalogBatchEntry[]): BatchRequest {
78 1 : return { scripts: entries.map(entry => ({
79 1 : id: entryIdOf(entry).toString(),
80 1 : text: entry.text,
81 1 : outputs: ['catalogDescriptor'],
82 1 : })) };
83 1 : }
84 :
85 1 : function commitCatalogBatch(catalog: DashQLCatalog, entries: CatalogBatchEntry[], result: BatchResult, validate: () => void) {
86 1 : validate();
87 1 : if (result.scripts.length !== entries.length) throw new Error('Catalog batch returned an unexpected script count');
88 1 : const prepared = entries.map(entry => {
89 1 : const matches = result.scripts.filter(script => script.id === entryIdOf(entry).toString());
90 1 : const output = matches[0];
91 1 : if (matches.length !== 1 || output.failure) {
92 1 : throw new Error(output?.failure ?? 'Catalog batch did not return a descriptor');
93 : }
94 1 : const error = output.diagnostics.find(diagnostic => diagnostic.severity === 'error');
95 1 : if (error) throw new Error(`Catalog contains invalid SQL: ${error.message}`);
96 1 : if (!output.catalogDescriptor?.length) throw new Error('Catalog batch did not return a descriptor');
97 1 : const descriptor = buffers.catalog.CatalogDescriptor.getRootAsCatalogDescriptor(new flatbuffers.ByteBuffer(output.catalogDescriptor));
98 1 : const counts = { tables: descriptor.tablesLength(), functions: descriptor.functionDeclarationsLength() };
99 1 : if (entry.requireFunctions && counts.functions === 0) throw new Error('Hyper function catalog is empty');
100 1 : return { descriptor: output.catalogDescriptor, counts };
101 1 : });
102 :
103 1 : validate();
104 1 : for (let i = 0; i < entries.length; ++i) {
105 1 : const entry = entries[i];
106 1 : const previousText = entry.script?.toString();
107 1 : try {
108 1 : entry.script?.replaceText(entry.text);
109 1 : catalog.replaceDescriptor(entryIdOf(entry), buffers.catalog.CatalogLayer.CONNECTION, prepared[i].descriptor);
110 1 : } catch (error) {
111 1 : try {
112 1 : if (previousText != null) entry.script?.replaceText(previousText);
113 1 : } catch {
114 : // Preserve the import/write failure if restoring this source also fails.
115 1 : }
116 1 : throw error;
117 1 : }
118 1 : }
119 1 : return prepared.map(entry => entry.counts);
120 1 : }
121 :
122 1 : export async function publishCatalogBatch(
123 1 : core: DashQL,
124 1 : catalog: DashQLCatalog,
125 1 : entries: CatalogBatchEntry[],
126 1 : options: CatalogBatchOptions = {},
127 1 : validate = beginCatalogBatch(catalog, options),
128 1 : ) {
129 1 : validate();
130 1 : if (entries.length === 0) return [];
131 1 : let lifetime = lifetimes.get(core);
132 1 : if (!lifetime) {
133 1 : lifetime = new AbortController();
134 1 : lifetimes.set(core, lifetime);
135 1 : }
136 1 : const validateBatch = validate;
137 1 : validate = () => {
138 1 : lifetime.signal.throwIfAborted();
139 1 : validateBatch();
140 1 : };
141 1 : validate();
142 1 : const processor = options.processor ?? ((request: BatchRequest) => getCatalogBatchWorker(core).processBatch(request));
143 1 : let onAbort: (() => void) | undefined;
144 1 : try {
145 1 : const cancelled = new Promise<never>((_resolve, reject) => {
146 1 : onAbort = () => reject(new DOMException('Catalog refresh was cancelled', 'AbortError'));
147 1 : options.abortSignal?.addEventListener('abort', onAbort, { once: true });
148 1 : lifetime.signal.addEventListener('abort', onAbort, { once: true });
149 1 : if (options.abortSignal?.aborted) onAbort();
150 1 : });
151 1 : const processing = new Promise<BatchResult>(resolve => resolve(processor(batchRequest(entries))));
152 1 : const result = await Promise.race([processing, cancelled]);
153 1 : validate();
154 1 : return commitCatalogBatch(catalog, entries, result, validate);
155 1 : } catch (error) {
156 : // A late worker failure from a disposed/superseded refresh is cancellation, not a
157 : // failure of the newer catalog. Promise.race also consumes late worker rejections.
158 1 : validate();
159 1 : throw error;
160 1 : } finally {
161 1 : if (onAbort) {
162 1 : options.abortSignal?.removeEventListener('abort', onAbort);
163 1 : lifetime.signal.removeEventListener('abort', onAbort);
164 1 : }
165 1 : }
166 1 : }
167 :
168 : /// Synchronous callers (initial prefetched functions and storage reconciliation) still prepare
169 : /// isolated batch descriptors, never scan/parse/analyze the persistence scripts on the main catalog.
170 1 : export function publishCatalogBatchSync(core: DashQL, catalog: DashQLCatalog, entries: CatalogBatchEntry[]) {
171 1 : const validate = beginCatalogBatch(catalog);
172 1 : if (entries.length === 0) return [];
173 1 : return commitCatalogBatch(catalog, entries, core.processBatch(batchRequest(entries)), validate);
174 1 : }
|
