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