Line data Source code
1 : import type { DashQL } from '@ankoh/dashql-core';
2 : import type { Logger } from '../../platform/logger/logger.js';
3 : import { stringifyError } from '../../platform/logger/logger.js';
4 : import { ProgressCounter } from '../../utils/progress.js';
5 : import type { ConnectionState } from '../../connections/connection_state.js';
6 : import { registerConnection, sessionConnectionFromState, type ConnectionRegistry } from '../../connections/connection_registry.js';
7 : import type { SessionScripts, ScriptData } from '../scripts/session_scripts.js';
8 : import { addSessionScriptsForConnection, copySessionScriptsRegistry, type SessionScriptsRegistry } from '../scripts/session_scripts_registry.js';
9 : import { createEmptyScriptData, destroySessionScripts, isEphemeralSession, replaceScriptSessionText, scriptConnectionOf, sessionConnectionIds } from '../scripts/session_scripts.js';
10 : import { readSessionAttachments, type SessionAttachment } from '../attachments/session_attachments.js';
11 : import { connectionStateFromStorage, destroyUnregisteredConnection, restoreConnectionCatalog } from '../../connections/connection_restore.js';
12 : import { CONNECTOR_INFOS, CONNECTOR_TYPES, ConnectorType } from '../../connections/connector_info.js';
13 : import type { StorageBackend, SessionEntry, SessionData, SessionRegistryBackend, ConnectionEntry, SessionConnection, SessionScript } from '../../storage/storage_backend.js';
14 : import { StorageBackendType, isSessionRegistryBackend } from '../../storage/storage_backend.js';
15 : import { connectionParamsEqual, createDefaultConnectionParamsForConnector } from '../../connections/connection_params.js';
16 : import { findEmbeddedConnection } from '../../connections/embedded_connection.js';
17 : import { patchSessionConnection, upsertConnectionEntry } from '../../storage/storage_writer.js';
18 : import { sanitizeSessionFiles, sessionConnectionIdsOf, withSessionScripts } from '../../storage/session_files.js';
19 : import { resolveSessionScripts, sessionScriptsResolutionChanged, type SessionConnectionResolver } from './session_script_connections.js';
20 : import { validateSessionData, describeInvalidSession, isLocalHyperWasmParams, isValidUuid, SessionValidationError, type InvalidSession } from '../../storage/session_validation.js';
21 : import type { BatchProcessor } from '../../connections/catalog/catalog_batch.js';
22 :
23 1 : const LOG_CTX = "app_state_loader";
24 :
25 : export interface RestoredAppState {
26 : connectionStates: Map<string, ConnectionState>;
27 : connectionStatesByType: string[][];
28 : sessionScripts: Map<string, SessionScripts>;
29 : sessionScriptsByConnection: Map<string, string[]>;
30 : /// Sessions whose metadata failed validation and were refused a load (keyed by bare UUID).
31 : /// These never enter the connection/session maps; the session workbench surfaces them as
32 : /// invalid (blocked from opening, still deletable).
33 : invalidSessions: Map<string, InvalidSession>;
34 : }
35 :
36 : /// Thrown by `restoreSessionEntry` when a session's metadata fails the up-front validation gate.
37 : ///
38 : /// Distinguished from an arbitrary restore error so the loader can record it as a (skipped)
39 : /// invalid session and surface it in the UI, rather than counting it as a hard failure.
40 1 : class InvalidSessionError extends Error {
41 1 : constructor(public readonly invalid: InvalidSession) {
42 1 : super(`invalid session ${invalid.sessionId}: ${invalid.error}`);
43 1 : this.name = 'InvalidSessionError';
44 1 : }
45 1 : }
46 :
47 : export interface AppStateRestorationProgress {
48 : restoreConnections: ProgressCounter;
49 : restoreCatalogs: ProgressCounter;
50 : restoreSessionScripts: ProgressCounter;
51 : }
52 :
53 : /// The progress counters of a restoration, every change is reported to the consumer
54 1 : class RestoreProgress {
55 : /// The restored connections of the sessions
56 1 : readonly connections = new ProgressCounter();
57 : /// The restored catalogs of the sessions
58 1 : readonly catalogs = new ProgressCounter();
59 : /// The restored session scripts
60 1 : readonly sessionScripts = new ProgressCounter();
61 :
62 1 : constructor(private readonly consumer: (progress: AppStateRestorationProgress) => void) { }
63 :
64 : /// Report the current progress
65 1 : report(): void {
66 1 : this.consumer({
67 1 : restoreConnections: this.connections.clone(),
68 1 : restoreCatalogs: this.catalogs.clone(),
69 1 : restoreSessionScripts: this.sessionScripts.clone(),
70 1 : });
71 1 : }
72 1 : }
73 :
74 : /// Restores session scripts from storage
75 1 : async function restoreSessionScripts(
76 1 : core: DashQL,
77 1 : backend: StorageBackend,
78 1 : sessionId: string,
79 1 : name: string | null,
80 1 : sessionData: SessionData,
81 1 : fallbackConnection: () => ConnectionState,
82 1 : resolveConnection: SessionConnectionResolver,
83 1 : logger: Logger
84 1 : ): Promise<{ sessionScripts: SessionScripts; session: SessionData }> {
85 1 : const scripts: Record<number, ScriptData> = {};
86 :
87 1 : try {
88 1 : logger.info("Loading scripts", { sessionId }, LOG_CTX);
89 1 : const storedScripts = await backend.loadScripts(sessionId);
90 1 : logger.info("Scripts loaded", {
91 1 : sessionId,
92 1 : scriptCount: storedScripts.length.toString()
93 1 : }, LOG_CTX);
94 :
95 : // The session file orders the scripts and names their connections
96 1 : const sqlByName = new Map(storedScripts.map(script => [script.name, script.sql]));
97 1 : const resolved = resolveSessionScripts(sessionData, storedScripts.map(script => script.name), fallbackConnection, resolveConnection,
98 1 : connectionId => logger.warn("Script references an unknown connection, using the embedded database", { sessionId, connectionId }, LOG_CTX));
99 1 : const scriptRefs: { [fileName: string]: { scriptId: number; fileName: string } } = {};
100 1 : for (const entry of resolved.scripts) {
101 1 : const [scriptKey, created] = createEmptyScriptData(core, scriptConnectionOf(entry.connection), entry.name);
102 1 : const scriptData = entry.unresolvedConnectionId == null ? created : { ...created, unresolvedConnectionId: entry.unresolvedConnectionId };
103 1 : scripts[scriptKey] = scriptData;
104 :
105 : // Set SQL content. Ordinary session scripts remain outdated until first use.
106 1 : replaceScriptSessionText(scriptData.scriptSession, sqlByName.get(entry.name) ?? '');
107 :
108 1 : scriptRefs[entry.name] = {
109 1 : scriptId: scriptKey,
110 1 : fileName: entry.name,
111 1 : };
112 1 : }
113 1 : const scriptOrder = resolved.scripts.map(entry => entry.name);
114 : // The attachments of the session, a Parquet attachment is a table parameter of the scripts
115 1 : let attachments: SessionAttachment[] = [];
116 1 : try {
117 1 : attachments = readSessionAttachments(await backend.listDataFiles(sessionId));
118 1 : } catch (error) {
119 1 : logger.warn("Listing the attachments failed", { sessionId, error: stringifyError(error) }, LOG_CTX);
120 1 : }
121 :
122 : const sessionScripts: SessionScripts = {
123 1 : instance: core,
124 1 : sessionId,
125 1 : name,
126 1 : pendingConnection: resolved.pendingConnection == null ? null : scriptConnectionOf(resolved.pendingConnection),
127 1 : sessionMetadata: sessionData.metadata,
128 1 : scripts,
129 1 : scriptRefs,
130 1 : scriptOrder,
131 1 : scriptFocus: { fileName: scriptOrder[0] ?? '' },
132 1 : semanticUserFocus: null,
133 1 : attachments,
134 : };
135 :
136 1 : return { sessionScripts, session: resolved.session };
137 1 : } catch (error) {
138 1 : for (const scriptData of Object.values(scripts)) {
139 1 : scriptData.scriptSession.destroy();
140 : }
141 1 : throw error;
142 1 : }
143 1 : }
144 :
145 : /// The state that the restoration of connections and sessions shares.
146 : ///
147 : /// Connections are entities of their own, the app manifest is their source of truth. They are
148 : /// restored first, sessions then resolve their references against them. Sessions keep copies of
149 : /// the connection params to stay self-contained, a copy that disagrees with the manifest is
150 : /// rewritten, and a reference to an unknown connection registers the connection from the copy.
151 : interface RestoreContext {
152 : /// The core instance
153 : core: DashQL;
154 : /// The storage backend
155 : backend: StorageBackend;
156 : /// The backend as registry, null if the backend doesn't own the connections
157 : registry: SessionRegistryBackend | null;
158 : /// The logger
159 : logger: Logger;
160 : /// The batch processor for catalog scripts
161 : processor?: BatchProcessor;
162 : /// The connections (connectionId -> state)
163 : connectionStates: Map<string, ConnectionState>;
164 : /// The connections by connector type
165 : connectionStatesByType: string[][];
166 : /// The connections whose catalog is restored or live
167 : catalogsRestored: Set<string>;
168 : /// The connections that were registered while restoring sessions
169 : newConnections: Map<string, ConnectionState>;
170 : /// The writes that bring the stored state in line with the restored one
171 : pendingWrites: RestorePendingWrites;
172 : }
173 :
174 : /// The writes that bring the stored state in line with the restored one
175 : interface RestorePendingWrites {
176 : /// Connections that sessions registered, they are missing in the app manifest
177 : connectionEntries: ConnectionEntry[];
178 : /// Catalog files of connections that hold dropped relations or changes, they are cleared
179 : staleConnectionFiles: Array<{ connectionId: string; fileName: string }>;
180 : /// Sessions whose copies of connection params are stale
181 : sessionCopies: Array<{ sessionId: string; connectionId: string }>;
182 : /// Sessions whose script lists or connections disagree with the restored scripts
183 : sessionScriptLists: Array<{ sessionId: string; scripts: SessionScript[]; connections: SessionConnection[] }>;
184 : /// Ephemeral sessions that hold scripts, the app stopped before their promotion was written
185 : promotedSessions: string[];
186 : }
187 :
188 1 : function createRestoreContext(core: DashQL, backend: StorageBackend, logger: Logger, processor: BatchProcessor | undefined): RestoreContext {
189 1 : return {
190 1 : core,
191 1 : backend,
192 1 : registry: isSessionRegistryBackend(backend) ? backend : null,
193 1 : logger,
194 1 : processor,
195 1 : connectionStates: new Map(),
196 1 : connectionStatesByType: CONNECTOR_TYPES.map(() => []),
197 1 : catalogsRestored: new Set(),
198 1 : newConnections: new Map(),
199 1 : pendingWrites: { connectionEntries: [], staleConnectionFiles: [], sessionCopies: [], sessionScriptLists: [], promotedSessions: [] },
200 1 : };
201 1 : }
202 :
203 : /// Restore a connection of the app manifest
204 1 : async function restoreConnectionEntry(ctx: RestoreContext, entry: ConnectionEntry): Promise<void> {
205 1 : let state: ConnectionState;
206 1 : try {
207 1 : state = connectionStateFromStorage(ctx.core, entry.id, entry);
208 1 : } catch (error) {
209 : // Sessions referencing the connection register it again from their copies
210 1 : ctx.logger.warn("Failed to restore connection", { connectionId: entry.id, error: stringifyError(error) }, LOG_CTX);
211 1 : return;
212 1 : }
213 1 : ctx.connectionStates.set(state.connectionId, state);
214 1 : ctx.connectionStatesByType[state.connectorInfo.connectorType].push(state.connectionId);
215 1 : await restoreCatalog(ctx, ctx.registry!, state, { connectionId: entry.id });
216 1 : }
217 :
218 : /// Restore the catalog of a connection once, returns false if the restoration failed
219 1 : async function restoreCatalog(
220 1 : ctx: RestoreContext,
221 1 : registry: SessionRegistryBackend,
222 1 : state: ConnectionState,
223 1 : logContext: Record<string, string>,
224 1 : ): Promise<boolean> {
225 1 : const connectionId = state.connectionId;
226 1 : ctx.catalogsRestored.add(connectionId);
227 1 : const restored = await restoreConnectionCatalog(ctx, registry, state, logContext);
228 1 : for (const fileName of restored.staleFiles) {
229 1 : ctx.pendingWrites.staleConnectionFiles.push({ connectionId, fileName });
230 : }
231 1 : return restored.succeeded;
232 1 : }
233 :
234 : /// Resolve a session's reference to a connection.
235 : /// The app manifest wins over the copy in the session, an unknown connection is registered from the copy.
236 : /// The app holds a single connection to the embedded database, an unknown one resolves to it.
237 1 : function resolveSessionConnection(ctx: RestoreContext, sessionId: string, database: SessionConnection): ConnectionState {
238 1 : const connectionId = database.connectionId;
239 1 : const known = ctx.connectionStates.get(connectionId)
240 1 : ?? (isLocalHyperWasmParams(database.params) ? findEmbeddedConnection(ctx.connectionStates.values()) : null);
241 1 : if (known != null) {
242 1 : if (known.connectionId === connectionId
243 1 : && (database.paramsDraft != null || database.name != null || !connectionParamsEqual(database.params, known.params))) {
244 1 : ctx.pendingWrites.sessionCopies.push({ sessionId, connectionId });
245 : }
246 1 : return known;
247 1 : }
248 1 : const state = connectionStateFromStorage(ctx.core, connectionId, database);
249 1 : ctx.connectionStates.set(state.connectionId, state);
250 1 : ctx.connectionStatesByType[state.connectorInfo.connectorType].push(state.connectionId);
251 1 : ctx.newConnections.set(state.connectionId, state);
252 1 : ctx.pendingWrites.connectionEntries.push({
253 1 : id: state.connectionId,
254 1 : params: state.params,
255 1 : ...(database.paramsDraft == null ? {} : { paramsDraft: database.paramsDraft }),
256 1 : ...(state.name == null ? {} : { name: state.name }),
257 1 : createdAt: new Date().toISOString(),
258 1 : });
259 1 : if (database.paramsDraft != null || database.name != null) {
260 : // Drafts and the names of imported connections move into the app manifest
261 1 : ctx.pendingWrites.sessionCopies.push({ sessionId, connectionId });
262 : }
263 1 : return state;
264 1 : }
265 :
266 : /// The connection to the embedded database, registered from a copy of the session or anew when the app
267 : /// has none. Scripts whose connections are unknown run on it.
268 1 : function resolveEmbeddedConnection(ctx: RestoreContext, sessionId: string, copies: readonly SessionConnection[]): ConnectionState {
269 1 : return findEmbeddedConnection(ctx.connectionStates.values()) ?? resolveSessionConnection(ctx, sessionId,
270 1 : copies.find(copy => isLocalHyperWasmParams(copy.params)) ?? {
271 0 : connectionId: crypto.randomUUID(),
272 0 : params: createDefaultConnectionParamsForConnector(CONNECTOR_INFOS[ConnectorType.HYPER]),
273 1 : });
274 1 : }
275 :
276 : /// Restores a single session (connection references + catalog + session).
277 : /// Returns the session as invalid when its scripts could not be restored.
278 1 : async function restoreSessionEntry(
279 1 : ctx: RestoreContext,
280 1 : sessionEntry: SessionEntry,
281 1 : sessionScripts: Map<string, SessionScripts>,
282 1 : sessionScriptsByConnection: Map<string, string[]>,
283 1 : progress: RestoreProgress,
284 1 : ): Promise<InvalidSession | null> {
285 1 : const { backend, logger } = ctx;
286 : // The session UUID is the authoritative identity and the key the backend routes on. Gate it up
287 : // front: a manifest entry whose path is not a valid UUID can't be loaded (the backend would
288 : // build a bogus path and throw), so surface it as an invalid session rather than a hard failure.
289 1 : const sessionId = sessionEntry.path;
290 1 : if (!isValidUuid(sessionId)) {
291 1 : const invalid = describeInvalidSession(sessionEntry, SessionValidationError.InvalidSessionId, null);
292 1 : logger.warn("Refusing to load session with an invalid id", {
293 1 : sessionId,
294 1 : reason: invalid.error,
295 1 : }, LOG_CTX);
296 1 : throw new InvalidSessionError(invalid);
297 1 : }
298 :
299 : // Phase 1: Resolve the connections
300 1 : logger.info("Restoring connection", { sessionId }, LOG_CTX);
301 1 : const connectionStartTime = performance.now();
302 1 : progress.connections.addStarted();
303 1 : progress.report();
304 :
305 1 : const sessionData = await loadValidSession(backend, logger, sessionEntry);
306 :
307 : // Validation above guarantees at least one connection and params that map to a known connector.
308 : // The connections of the session are the connections of its scripts, copies that no script uses
309 : // are not registered.
310 1 : const used = new Set(sessionConnectionIdsOf(sanitizeSessionFiles(sessionData).scripts, sessionData.connections[0].connectionId));
311 1 : const usedCopies = sessionData.connections.filter(copy => used.has(copy.connectionId));
312 1 : const connectionStates = usedCopies.length > 0
313 1 : ? usedCopies.map(copy => resolveSessionConnection(ctx, sessionId, copy))
314 1 : : [resolveEmbeddedConnection(ctx, sessionId, sessionData.connections)];
315 1 : const connectorInfo = connectionStates[0].connectorInfo;
316 1 : const resolveScriptConnection: SessionConnectionResolver = (connectionId, copy) => copy != null
317 1 : ? resolveSessionConnection(ctx, sessionId, copy)
318 1 : : ctx.connectionStates.get(connectionId) ?? null;
319 :
320 1 : const connectionDuration = performance.now() - connectionStartTime;
321 1 : logger.info("Connections restored", {
322 1 : sessionId,
323 1 : connectorTypes: connectionStates.map(state => ConnectorType[state.connectorInfo.connectorType]).join(','),
324 1 : durationMs: connectionDuration.toFixed(2)
325 1 : }, LOG_CTX);
326 :
327 1 : progress.connections.addSucceeded();
328 :
329 : // Phase 2: Restore the catalogs of the connections that the session registered. The connections
330 : // of the app manifest restored their catalogs already.
331 1 : logger.info("Restoring catalog", { sessionId }, LOG_CTX);
332 1 : progress.catalogs.addStarted();
333 1 : progress.report();
334 1 : let catalogsRestored = true;
335 1 : for (const connectionState of connectionStates) {
336 1 : if (ctx.registry == null || ctx.catalogsRestored.has(connectionState.connectionId)) continue;
337 1 : catalogsRestored = await restoreCatalog(ctx, ctx.registry, connectionState, { sessionId, connectionId: connectionState.connectionId }) && catalogsRestored;
338 1 : }
339 1 : if (catalogsRestored) {
340 1 : progress.catalogs.addSucceeded();
341 : } else {
342 1 : progress.catalogs.addFailed();
343 : }
344 :
345 : // Phase 3: Restore session scripts
346 1 : logger.info("Restoring session scripts", { sessionId }, LOG_CTX);
347 1 : const sessionScriptsStartTime = performance.now();
348 1 : progress.sessionScripts.addStarted();
349 1 : progress.report();
350 :
351 1 : let restoredSessionScripts: SessionScripts | null = null;
352 1 : try {
353 1 : const restored = await restoreSessionScripts(
354 1 : ctx.core,
355 1 : backend,
356 1 : sessionId,
357 1 : sessionData.name?.trim() || null,
358 1 : sessionData,
359 1 : () => resolveEmbeddedConnection(ctx, sessionId, sessionData.connections),
360 1 : resolveScriptConnection,
361 1 : logger
362 1 : );
363 1 : restoredSessionScripts = restored.sessionScripts;
364 1 : if (isEphemeralSession(restoredSessionScripts) && restoredSessionScripts.scriptOrder.length > 0) {
365 1 : const { ephemeral: _promoted, ...sessionMetadata } = restoredSessionScripts.sessionMetadata;
366 1 : restoredSessionScripts = { ...restoredSessionScripts, sessionMetadata };
367 1 : ctx.pendingWrites.promotedSessions.push(sessionId);
368 1 : }
369 1 : if (sessionScriptsResolutionChanged(sessionData, restored.session)) {
370 1 : const { scripts, connections } = restored.session;
371 1 : ctx.pendingWrites.sessionScriptLists.push({ sessionId, scripts, connections });
372 1 : }
373 :
374 1 : const sessionScriptsDuration = performance.now() - sessionScriptsStartTime;
375 1 : logger.info("Session scripts restored", {
376 1 : sessionId,
377 1 : scriptCount: Object.keys(restoredSessionScripts.scripts).length.toString(),
378 1 : durationMs: sessionScriptsDuration.toFixed(2)
379 1 : }, LOG_CTX);
380 :
381 1 : progress.sessionScripts.addSucceeded();
382 1 : } catch (sessionScriptsError) {
383 1 : const sessionScriptsDuration = performance.now() - sessionScriptsStartTime;
384 :
385 1 : logger.warn("Failed to restore session scripts", {
386 1 : sessionId,
387 1 : durationMs: sessionScriptsDuration.toFixed(2),
388 1 : error: stringifyError(sessionScriptsError),
389 1 : stack: (sessionScriptsError instanceof Error ? sessionScriptsError.stack : (sessionScriptsError as any)?.stack)?.substring(0, 500)
390 1 : }, LOG_CTX);
391 :
392 1 : progress.sessionScripts.addFailed();
393 1 : }
394 :
395 1 : progress.report();
396 1 : if (restoredSessionScripts == null) {
397 1 : return describeInvalidSession(sessionEntry, SessionValidationError.SessionNotRestored, sessionData);
398 : }
399 1 : sessionScripts.set(sessionId, restoredSessionScripts);
400 1 : for (const id of sessionConnectionIds(restoredSessionScripts)) {
401 1 : sessionScriptsByConnection.set(id, [...(sessionScriptsByConnection.get(id) ?? []), sessionId]);
402 : }
403 1 : return null;
404 1 : }
405 :
406 : /// Load a session and validate its metadata
407 1 : async function loadValidSession(backend: StorageBackend, logger: Logger, sessionEntry: SessionEntry): Promise<SessionData> {
408 1 : const sessionId = sessionEntry.path;
409 1 : logger.info("Loading session data", { sessionId }, LOG_CTX);
410 : // The manifest registers this session, but its files may be gone (a native session whose folder
411 : // the user moved/deleted, or a corrupt/absent OPFS session). Treat a load failure the same as a
412 : // metadata rejection: surface it as an invalid session so the selector shows it blocked and
413 : // deletable, rather than letting it fall through to the loader's generic "failed to restore"
414 : // path — which logs an error on every launch and leaves the stale entry with no way to remove it.
415 1 : let sessionData: SessionData;
416 1 : try {
417 1 : sessionData = await backend.loadSession(sessionId);
418 1 : } catch (loadError) {
419 1 : const invalid = describeInvalidSession(sessionEntry, SessionValidationError.SessionUnreadable, null);
420 1 : logger.warn("Refusing to load session with unreadable files", {
421 1 : sessionId,
422 1 : reason: invalid.error,
423 1 : error: stringifyError(loadError),
424 1 : }, LOG_CTX);
425 1 : throw new InvalidSessionError(invalid);
426 1 : }
427 :
428 : // Fail-fast metadata validation: refuse to load a session whose metadata is structurally
429 : // unusable (unsupported format, no id, or invalid connection). This runs
430 : // before any heavy restore work and surfaces the session as invalid in the selector rather than
431 : // letting it blow up mid-restore. Runtime hiccups (catalog/session) remain non-fatal below.
432 1 : const validation = validateSessionData(sessionData);
433 1 : if (!validation.ok) {
434 1 : const invalid = describeInvalidSession(sessionEntry, validation.error, sessionData);
435 1 : logger.warn("Refusing to load invalid session", {
436 1 : sessionId,
437 1 : reason: invalid.error,
438 1 : }, LOG_CTX);
439 1 : throw new InvalidSessionError(invalid);
440 1 : }
441 1 : logger.info("Session data loaded", { sessionId }, LOG_CTX);
442 1 : return sessionData;
443 1 : }
444 :
445 : /// Perform the writes that bring the stored state in line with the restored one.
446 : /// Failures are logged, the next load attempts them again.
447 : ///
448 : /// The writes go to the backend directly, not through the storage writer. They run before the restored
449 : /// sessions join the registry, so no write of the storage writer touches these sessions yet, and the
450 : /// backend serializes the updates of the app manifest.
451 1 : async function flushRestoreWrites(ctx: RestoreContext): Promise<void> {
452 1 : const { registry, pendingWrites } = ctx;
453 1 : if (registry != null) {
454 1 : await flushRestoredConnectionWrites(ctx, registry);
455 : }
456 1 : for (const { sessionId, scripts, connections } of pendingWrites.sessionScriptLists) {
457 1 : await patchRestoredSession(ctx, sessionId, session => {
458 1 : const next = withSessionScripts(session, scripts, connections);
459 1 : return sessionScriptsResolutionChanged(session, next) ? next : null;
460 1 : });
461 : }
462 1 : for (const sessionId of pendingWrites.promotedSessions) {
463 1 : await patchRestoredSession(ctx, sessionId, session => {
464 1 : const { ephemeral: _promoted, ...metadata } = session.metadata;
465 1 : return { ...session, metadata };
466 1 : });
467 : }
468 1 : }
469 :
470 : /// Patch the file of a restored session, a failure is logged and the next load attempts it again
471 1 : async function patchRestoredSession(ctx: RestoreContext, sessionId: string, patch: (session: SessionData) => SessionData | null): Promise<void> {
472 1 : try {
473 1 : const next = patch(await ctx.backend.loadSession(sessionId));
474 1 : if (next != null) {
475 1 : await ctx.backend.saveSessionManifest(sessionId, next);
476 : }
477 1 : } catch (error) {
478 0 : ctx.logger.warn("Failed to update a restored session", { sessionId, error: stringifyError(error) }, LOG_CTX);
479 0 : }
480 1 : }
481 :
482 : /// Persist the connections that the restoration registered
483 1 : async function flushRestoredConnectionWrites(ctx: RestoreContext, registry: SessionRegistryBackend): Promise<void> {
484 1 : const { logger, pendingWrites } = ctx;
485 1 : try {
486 1 : if (pendingWrites.connectionEntries.length > 0) {
487 1 : await registry.updateConnections(connections => {
488 1 : let next = connections ?? [];
489 1 : for (const entry of pendingWrites.connectionEntries) {
490 1 : next = upsertConnectionEntry(next, entry) ?? next;
491 : }
492 1 : return next;
493 1 : });
494 : }
495 1 : for (const { connectionId, fileName } of pendingWrites.staleConnectionFiles) {
496 1 : await registry.saveConnectionFile(connectionId, fileName, '');
497 : }
498 1 : } catch (error) {
499 1 : logger.warn("Failed to persist restored connections", { error: stringifyError(error) }, LOG_CTX);
500 1 : return;
501 1 : }
502 1 : for (const { sessionId, connectionId } of pendingWrites.sessionCopies) {
503 1 : const connection = ctx.connectionStates.get(connectionId);
504 1 : if (connection == null) continue;
505 1 : try {
506 1 : const existing = await ctx.backend.loadSession(sessionId);
507 1 : const next = patchSessionConnection(existing, sessionConnectionFromState(connection));
508 1 : if (next != null) {
509 1 : await ctx.backend.saveSessionManifest(sessionId, next);
510 : }
511 1 : } catch (error) {
512 1 : logger.warn("Failed to update the connection copy of a session", { sessionId, connectionId, error: stringifyError(error) }, LOG_CTX);
513 1 : }
514 1 : }
515 1 : }
516 :
517 : /// The connection references + session a single session restored into.
518 : export interface RestoredSession {
519 : /// The session id
520 : sessionId: string;
521 : /// The connections the session registered, connections that were already live are not included
522 : databases: ConnectionState[];
523 : /// The session scripts
524 : sessionScripts: SessionScripts;
525 : }
526 :
527 1 : export function mergeRestoredSessionIntoConnections(
528 1 : reg: ConnectionRegistry,
529 1 : restored: RestoredSession,
530 1 : ): ConnectionRegistry {
531 1 : for (const database of restored.databases) {
532 1 : if (!reg.connections.has(database.connectionId)) {
533 1 : registerConnection(reg, database);
534 : }
535 : }
536 1 : return { ...reg };
537 1 : }
538 :
539 1 : export function mergeRestoredSessionIntoScripts(
540 1 : reg: SessionScriptsRegistry,
541 1 : restored: RestoredSession,
542 1 : ): SessionScriptsRegistry {
543 1 : const next = copySessionScriptsRegistry(reg);
544 1 : next.sessionScriptsMap.set(restored.sessionId, restored.sessionScripts);
545 1 : for (const connectionId of sessionConnectionIds(restored.sessionScripts)) {
546 1 : addSessionScriptsForConnection(next, connectionId, restored.sessionId);
547 : }
548 1 : return next;
549 1 : }
550 :
551 1 : export function destroyRestoredSession(restored: RestoredSession): void {
552 1 : destroySessionScripts(restored.sessionScripts);
553 1 : for (const database of restored.databases) destroyUnregisteredConnection(database);
554 1 : }
555 :
556 : /// The live connections that a restored session resolves its references against
557 : export type LiveConnections = Pick<ConnectionRegistry, 'connections'>;
558 :
559 : /// Restore a single, already-persisted session (connection references + catalog + session) and
560 : /// return just that session's pieces.
561 : ///
562 : /// This is the incremental counterpart to `restoreAppState`, used after a session is written to
563 : /// storage at runtime (e.g. imported from a shared URL) so it can be merged into the already-live
564 : /// registries without a full app reload. It reuses the exact same `restoreSessionEntry` path the boot
565 : /// loader runs, so a URL-imported session is decoded and cataloged identically to one
566 : /// loaded at startup. References to live connections reuse them, unknown connections are
567 : /// registered and returned.
568 1 : export async function restoreSingleSession(
569 1 : core: DashQL,
570 1 : backend: StorageBackend,
571 1 : logger: Logger,
572 1 : sessionId: string,
573 1 : live: LiveConnections,
574 1 : processor?: BatchProcessor,
575 1 : ): Promise<RestoredSession> {
576 : // A freshly imported session is implicitly OPFS-backed and keyed by its UUID; that's all the
577 : // manifest entry `restoreSessionEntry` needs to route the load.
578 1 : const sessionEntry: SessionEntry = { path: sessionId, storageType: StorageBackendType.OPFS };
579 :
580 1 : const ctx = createRestoreContext(core, backend, logger, processor);
581 1 : for (const [connectionId, state] of live.connections) {
582 1 : ctx.connectionStates.set(connectionId, state);
583 1 : ctx.catalogsRestored.add(connectionId);
584 1 : }
585 1 : const sessionScripts = new Map<string, SessionScripts>();
586 1 : const sessionScriptsByConnection = new Map<string, string[]>();
587 :
588 1 : try {
589 1 : await restoreSessionEntry(
590 1 : ctx,
591 1 : sessionEntry,
592 1 : sessionScripts,
593 1 : sessionScriptsByConnection,
594 1 : new RestoreProgress(() => { }),
595 1 : );
596 1 : } catch (error) {
597 1 : for (const database of ctx.newConnections.values()) destroyUnregisteredConnection(database);
598 1 : throw error;
599 1 : }
600 :
601 1 : const scripts = sessionScripts.get(sessionId) ?? null;
602 1 : if (!scripts) {
603 0 : for (const database of ctx.newConnections.values()) destroyUnregisteredConnection(database);
604 0 : throw new Error(`imported session ${sessionId} did not restore its scripts`);
605 0 : }
606 1 : await flushRestoreWrites(ctx);
607 1 : return {
608 1 : sessionId,
609 1 : databases: [...ctx.newConnections.values()],
610 1 : sessionScripts: scripts,
611 1 : };
612 1 : }
613 :
614 : /**
615 : * Restores complete application state from storage
616 : */
617 1 : export async function restoreAppState(
618 1 : core: DashQL,
619 1 : backend: StorageBackend,
620 1 : logger: Logger,
621 1 : progressConsumer: (progress: AppStateRestorationProgress) => void,
622 1 : processor?: BatchProcessor,
623 1 : ): Promise<RestoredAppState> {
624 1 : logger.info("Starting app state restoration", {}, LOG_CTX);
625 1 : const startTime = performance.now();
626 :
627 1 : const ctx = createRestoreContext(core, backend, logger, processor);
628 1 : const sessionScripts = new Map<string, SessionScripts>();
629 1 : const sessionScriptsByConnection = new Map<string, string[]>();
630 1 : const invalidSessions = new Map<string, InvalidSession>();
631 :
632 1 : const progress = new RestoreProgress(progressConsumer);
633 :
634 1 : try {
635 : // Load manifest
636 1 : logger.info("Loading app manifest", {}, LOG_CTX);
637 1 : const manifestStartTime = performance.now();
638 1 : const sessionEntries = await backend.listSessions();
639 1 : let connectionEntries: ConnectionEntry[] = [];
640 1 : if (ctx.registry != null) {
641 1 : const manifest = await ctx.registry.loadManifest();
642 1 : connectionEntries = manifest.connections ?? [];
643 : }
644 1 : const manifestDuration = performance.now() - manifestStartTime;
645 :
646 1 : logger.info("Loaded app manifest", {
647 1 : sessionCount: sessionEntries.length.toString(),
648 1 : connectionCount: connectionEntries.length.toString(),
649 1 : durationMs: manifestDuration.toFixed(2)
650 1 : }, LOG_CTX);
651 :
652 : // Restore the connections before the sessions that reference them
653 1 : for (const entry of connectionEntries) {
654 1 : await restoreConnectionEntry(ctx, entry);
655 : }
656 :
657 : // Set totals.
658 1 : progress.connections.addTotal(sessionEntries.length);
659 1 : progress.catalogs.addTotal(sessionEntries.length);
660 1 : progress.sessionScripts.addTotal(sessionEntries.length);
661 :
662 1 : progress.report();
663 :
664 : // Process each session
665 1 : logger.info("Restoring sessions", { count: sessionEntries.length.toString() }, LOG_CTX);
666 1 : for (let i = 0; i < sessionEntries.length; i++) {
667 1 : const sessionEntry = sessionEntries[i];
668 1 : const sessionStartTime = performance.now();
669 :
670 1 : try {
671 1 : logger.info("Restoring session", {
672 1 : index: `${i + 1}/${sessionEntries.length}`,
673 1 : sessionId: sessionEntry.path
674 1 : }, LOG_CTX);
675 :
676 1 : const invalid = await restoreSessionEntry(
677 1 : ctx,
678 1 : sessionEntry,
679 1 : sessionScripts,
680 1 : sessionScriptsByConnection,
681 1 : progress,
682 1 : );
683 1 : if (invalid != null) {
684 : // The session stays visible as invalid, the user can delete it
685 1 : invalidSessions.set(invalid.sessionId, invalid);
686 : }
687 :
688 1 : const sessionDuration = performance.now() - sessionStartTime;
689 1 : logger.info("Session restored", {
690 1 : index: `${i + 1}/${sessionEntries.length}`,
691 1 : sessionId: sessionEntry.path,
692 1 : durationMs: sessionDuration.toFixed(2)
693 1 : }, LOG_CTX);
694 1 : } catch (error) {
695 1 : const sessionDuration = performance.now() - sessionStartTime;
696 :
697 1 : if (error instanceof InvalidSessionError) {
698 : // Metadata validation refused this session up front. Record it so the selector
699 : // can show it as invalid (blocked, deletable), and account it as *skipped*
700 : // rather than *failed* — nothing was attempted, the metadata was simply
701 : // unusable. The session contributed no connection/catalog/scripts/scripts.
702 1 : invalidSessions.set(error.invalid.sessionId, error.invalid);
703 1 : progress.connections.addSkipped();
704 1 : progress.catalogs.addSkipped();
705 1 : progress.sessionScripts.addSkipped();
706 :
707 1 : progress.report();
708 1 : continue;
709 1 : }
710 :
711 1 : logger.warn("Failed to restore session", {
712 1 : index: `${i + 1}/${sessionEntries.length}`,
713 1 : sessionId: sessionEntry.path,
714 1 : durationMs: sessionDuration.toFixed(2),
715 1 : error: stringifyError(error)
716 1 : }, LOG_CTX);
717 :
718 : // The session stays visible as invalid, the user can delete it
719 1 : invalidSessions.set(sessionEntry.path, describeInvalidSession(sessionEntry, SessionValidationError.SessionNotRestored, null));
720 1 : progress.connections.addFailed();
721 1 : progress.catalogs.addFailed();
722 1 : progress.sessionScripts.addFailed();
723 1 : progress.report();
724 1 : }
725 1 : }
726 1 : } catch (manifestError) {
727 1 : logger.warn("Failed to load manifest, starting with empty state", {
728 1 : error: stringifyError(manifestError)
729 1 : }, LOG_CTX);
730 1 : }
731 :
732 1 : await flushRestoreWrites(ctx);
733 :
734 1 : progress.report();
735 :
736 1 : const totalDuration = performance.now() - startTime;
737 1 : logger.info("Finished loading app state", {
738 1 : connections: ctx.connectionStates.size.toString(),
739 1 : sessionScripts: sessionScripts.size.toString(),
740 1 : invalidSessions: invalidSessions.size.toString(),
741 1 : connectionsSucceeded: progress.connections.succeeded.toString(),
742 1 : connectionsFailed: progress.connections.failed.toString(),
743 1 : connectionsSkipped: progress.connections.skipped.toString(),
744 1 : totalDurationMs: totalDuration.toFixed(2)
745 1 : }, LOG_CTX);
746 :
747 1 : return {
748 1 : connectionStates: ctx.connectionStates,
749 1 : connectionStatesByType: ctx.connectionStatesByType,
750 1 : sessionScripts,
751 1 : sessionScriptsByConnection,
752 1 : invalidSessions,
753 1 : };
754 1 : }
|
