Line data Source code
1 : import type * as arrow from 'apache-arrow';
2 :
3 : import type { QueryResultCache, QueryResultCacheEntry } from '../query/query_result_cache.js';
4 : import { copyArrowTableAsParquet, type ParquetDatabase } from '../platform/hyperdb/hyperdb_parquet_export.js';
5 : import { getOPFSRoot } from '../platform/opfs_root.js';
6 : import { DEFAULT_CACHE_MAX_BYTES, DEFAULT_CACHE_MAX_FILES, evictToFit, type CacheFileStat, type QueryResultCacheStore } from './query_result_cache_eviction.js';
7 :
8 : /// The folder of the query result cache in the origin private file system
9 1 : export const QUERY_RESULT_CACHE_DIRECTORY = 'cache';
10 : /// The mount path of the query result cache in the compute database
11 1 : export const QUERY_RESULT_CACHE_MOUNT = '/mnt/external/cache';
12 : /// The empty marker beside a file whose modification time is the last read of the file
13 1 : const ACCESS_SUFFIX = '.last_access';
14 : /// The file of a key, `<key>.<uuid>.parquet`
15 1 : const KEY_FILE = /^([0-9a-f]{64})\.[0-9a-f-]{36}\.parquet$/;
16 : /// The suffix of a transient file, a result that can't be reused or a file that is being written. A transient file
17 : /// that nobody pins is garbage, every listing removes it.
18 1 : const TRANSIENT_SUFFIX = '.transient.parquet';
19 : /// The Web Lock of a file, readers hold it shared, eviction and deletion take it exclusively if nobody holds it
20 1 : const LOCK_PREFIX = 'dashql-query-result-cache/';
21 :
22 : /// The options of the query result cache
23 : export interface OPFSQueryResultCacheOptions {
24 : /// The origin private file system, the root of the files of the app
25 : getRoot?: () => Promise<FileSystemDirectoryHandle>;
26 : /// The locks of the windows of the app
27 : locks?: LockManager;
28 : /// The maximum total size of the files
29 : maxBytes?: number;
30 : /// The maximum number of files
31 : maxFiles?: number;
32 : }
33 :
34 : /// The key of a file, null for transient files and files of other versions
35 1 : export function cacheFileKey(name: string): string | null {
36 1 : return KEY_FILE.exec(name)?.[1] ?? null;
37 1 : }
38 :
39 : /// Is a file transient, i.e. removed once nobody pins it?
40 1 : export function isTransientCacheFile(name: string): boolean {
41 1 : return name.endsWith(TRANSIENT_SUFFIX);
42 1 : }
43 :
44 : /// The query result cache of the app in the folder `cache/` of the origin private file system
45 1 : export function createOPFSQueryResultCache(options: OPFSQueryResultCacheOptions = {}): QueryResultCache {
46 1 : const getRoot = options.getRoot ?? getOPFSRoot;
47 1 : const locks = options.locks ?? navigator.locks;
48 1 : const maxBytes = options.maxBytes ?? DEFAULT_CACHE_MAX_BYTES;
49 1 : const maxFiles = options.maxFiles ?? DEFAULT_CACHE_MAX_FILES;
50 1 : const directory = async () => await (await getRoot()).getDirectoryHandle(QUERY_RESULT_CACHE_DIRECTORY, { create: true });
51 :
52 : /// The Parquet files with their sizes and times. Transient files that nobody pins and markers without a file are
53 : /// removed.
54 1 : const listFiles = async (): Promise<CacheFileStat[]> => {
55 1 : const dir = await directory();
56 1 : const accessMs = new Map<string, number>();
57 1 : let files: { name: string; size: number; mtimeMs: number }[] = [];
58 1 : for await (const [name, handle] of dir.entries()) {
59 1 : if (handle.kind !== 'file') continue;
60 1 : try {
61 1 : const file = await (handle as FileSystemFileHandle).getFile();
62 1 : if (name.endsWith(ACCESS_SUFFIX)) accessMs.set(name.slice(0, -ACCESS_SUFFIX.length), file.lastModified);
63 1 : else if (name.endsWith('.parquet')) files.push({ name, size: file.size, mtimeMs: file.lastModified });
64 1 : } catch {
65 : // A file that another window removed meanwhile
66 0 : }
67 1 : }
68 : // The pinned transient files are written or read right now, they stay and count towards the limits. An empty
69 : // file of a key is being written without `move` or was left by a window that closed during the write: readers
70 : // find no result in it.
71 1 : const removed = new Set<string>();
72 1 : const incomplete = (file: { name: string; size: number }) => file.size === 0 && cacheFileKey(file.name) != null;
73 1 : for (const file of files) {
74 1 : if ((isTransientCacheFile(file.name) || incomplete(file)) && await tryRemove(file.name)) removed.add(file.name);
75 : }
76 1 : files = files.filter(file => !removed.has(file.name) && !incomplete(file));
77 1 : const names = new Set(files.map(file => file.name));
78 1 : for (const name of accessMs.keys()) {
79 1 : if (!names.has(name)) await dir.removeEntry(`${name}${ACCESS_SUFFIX}`).catch(() => {});
80 : }
81 1 : return files.map(file => ({ ...file, lastAccessMs: accessMs.get(file.name) ?? file.mtimeMs }));
82 1 : };
83 1 : const entries = async (): Promise<QueryResultCacheEntry[]> => (await listFiles()).flatMap(file => {
84 1 : const key = cacheFileKey(file.name);
85 1 : return key == null ? [] : [{ name: file.name, key, size: file.size, cachedAtMs: file.mtimeMs, lastAccessMs: file.lastAccessMs }];
86 1 : });
87 :
88 : // The compute database that the files are mounted in, shared by the readers and writers of this window
89 1 : let mountedIn: ParquetDatabase | null = null;
90 : // Did this window remove files since the mount was refreshed? The database still knows the files it opened.
91 1 : let mountStale = false;
92 1 : let mountChain: Promise<unknown> = Promise.resolve();
93 1 : const serialize = <T>(operation: () => Promise<T>): Promise<T> => {
94 1 : const result = mountChain.then(operation);
95 1 : mountChain = result.catch(() => {});
96 1 : return result;
97 1 : };
98 :
99 : /// Remove a file and its marker if nobody reads it, resolves true if the file is gone
100 1 : const tryRemove = async (name: string): Promise<boolean> => {
101 1 : return await locks.request(`${LOCK_PREFIX}${name}`, { mode: 'exclusive', ifAvailable: true }, async lock => {
102 1 : if (lock == null) return false;
103 1 : const dir = await directory();
104 1 : let gone = true;
105 1 : await dir.removeEntry(name).catch((error: unknown) => {
106 0 : if ((error as DOMException)?.name !== 'NotFoundError') gone = false;
107 1 : });
108 1 : if (!gone) return false;
109 1 : await dir.removeEntry(`${name}${ACCESS_SUFFIX}`).catch(() => {});
110 1 : mountStale = true;
111 1 : return true;
112 1 : });
113 1 : };
114 1 : const store: QueryResultCacheStore = {
115 1 : listCacheFiles: listFiles,
116 1 : deleteCacheFile: async name => await tryRemove(name),
117 1 : };
118 1 : const touch = async (name: string) => {
119 1 : const writable = await (await (await directory()).getFileHandle(`${name}${ACCESS_SUFFIX}`, { create: true })).createWritable();
120 1 : await writable.close();
121 1 : };
122 1 : const pin = (name: string): Promise<() => Promise<void>> => new Promise(acquired => {
123 : // The request settles once the lock is released, not when the callback returns
124 1 : const held = locks.request(`${LOCK_PREFIX}${name}`, { mode: 'shared' }, () => new Promise<void>(release => {
125 1 : acquired(async () => {
126 1 : release();
127 1 : await held.catch(() => {});
128 1 : });
129 1 : }));
130 1 : });
131 1 : const mount = (database: ParquetDatabase) => serialize(async () => {
132 1 : if (mountedIn != null && mountedIn !== database) {
133 1 : await mountedIn.removeExternalDirectory(QUERY_RESULT_CACHE_MOUNT).catch(() => {});
134 1 : mountedIn = null;
135 1 : }
136 1 : if (mountedIn == null) {
137 1 : mountStale = false;
138 1 : await database.registerExternalDirectory(await directory(), QUERY_RESULT_CACHE_MOUNT, 'readwrite');
139 1 : mountedIn = database;
140 1 : } else if (mountStale) {
141 : // Removals during the refresh mark the mount stale again
142 1 : mountStale = false;
143 1 : try {
144 1 : await database.refreshExternalDirectory(QUERY_RESULT_CACHE_MOUNT);
145 1 : } catch (error) {
146 1 : mountStale = true;
147 1 : throw error;
148 1 : }
149 1 : }
150 1 : });
151 :
152 1 : const lookup = async (key: string) => {
153 1 : const files = (await entries()).filter(entry => entry.key === key);
154 1 : files.sort((a, b) => b.cachedAtMs - a.cachedAtMs || b.name.localeCompare(a.name));
155 1 : return files[0] ?? null;
156 1 : };
157 :
158 1 : return {
159 1 : mountPath: QUERY_RESULT_CACHE_MOUNT,
160 1 : lookup,
161 1 : has: async key => (await entries()).some(entry => entry.key === key),
162 1 : retain: async key => {
163 : // Another window may remove the file between the lookup and the pin, then a newer file may be there
164 1 : for (let attempt = 0; attempt < 3; ++attempt) {
165 1 : const entry = await lookup(key);
166 1 : if (entry == null) return null;
167 1 : const release = await pin(entry.name);
168 1 : try {
169 1 : await (await directory()).getFileHandle(entry.name);
170 1 : } catch {
171 0 : await release();
172 0 : continue;
173 0 : }
174 1 : await touch(entry.name).catch(() => {});
175 1 : return { entry, release };
176 1 : }
177 0 : return null;
178 1 : },
179 1 : list: entries,
180 1 : touch,
181 1 : delete: async key => {
182 1 : let gone = true;
183 1 : for (const entry of (await entries()).filter(entry => entry.key === key)) {
184 1 : if (!await tryRemove(entry.name)) gone = false;
185 : }
186 1 : return gone;
187 1 : },
188 1 : clear: async () => {
189 : // Also the files without keys that nobody reads, e.g. those of runs of a window that was closed
190 1 : const keys = new Set<string>();
191 1 : for (const file of await listFiles()) {
192 1 : const key = cacheFileKey(file.name);
193 1 : if (await tryRemove(file.name) && key != null) keys.add(key);
194 1 : }
195 : // A key whose newer file is read stays cached
196 1 : const remaining = new Set((await entries()).map(entry => entry.key));
197 1 : return [...keys].filter(key => !remaining.has(key));
198 1 : },
199 1 : pin,
200 1 : newFileName: key => `${key}.${crypto.randomUUID()}.parquet`,
201 1 : newTransientFileName: () => `${crypto.randomUUID()}${TRANSIENT_SUFFIX}`,
202 1 : mount,
203 1 : writeTable: async (database: ParquetDatabase, table: arrow.Table, name: string) => {
204 1 : await mount(database);
205 1 : const dir = await directory();
206 1 : const key = cacheFileKey(name);
207 1 : if (key == null) {
208 0 : await copyArrowTableAsParquet(database, table, `${QUERY_RESULT_CACHE_MOUNT}/${name}`);
209 0 : return await (await dir.getFileHandle(name)).getFile();
210 0 : }
211 : // Readers of other windows find the file of a key once it is complete
212 1 : const temporary = `${crypto.randomUUID()}${TRANSIENT_SUFFIX}`;
213 1 : const releaseTemporary = await pin(temporary);
214 1 : try {
215 1 : await copyArrowTableAsParquet(database, table, `${QUERY_RESULT_CACHE_MOUNT}/${temporary}`);
216 1 : const handle = await dir.getFileHandle(temporary);
217 1 : if ('move' in handle && typeof (handle as { move?: unknown }).move === 'function') {
218 1 : await (handle as FileSystemFileHandle & { move(name: string): Promise<void> }).move(name);
219 0 : } else {
220 0 : await (await handle.getFile()).stream().pipeTo(await (await dir.getFileHandle(name, { create: true })).createWritable());
221 0 : await dir.removeEntry(temporary);
222 0 : }
223 1 : } finally {
224 1 : await releaseTemporary();
225 1 : }
226 1 : await touch(name);
227 : // The older files of the key are removed once nobody reads them, eviction keeps the files below the limits
228 1 : for (const entry of await entries()) {
229 1 : if (entry.key === key && entry.name !== name) await tryRemove(entry.name);
230 : }
231 1 : await evictToFit(store, maxBytes, maxFiles);
232 1 : return await (await dir.getFileHandle(name)).getFile();
233 1 : },
234 1 : readTable: async (database: ParquetDatabase, name: string) => {
235 1 : const release = await pin(name);
236 1 : try {
237 : // The file may have been removed between the lookup and the pin
238 1 : await (await directory()).getFileHandle(name);
239 1 : await mount(database);
240 1 : const connection = await database.connect({ pooled: true });
241 1 : try {
242 1 : return await connection.query(`SELECT * FROM external('${QUERY_RESULT_CACHE_MOUNT}/${name}', format => 'parquet')`);
243 1 : } finally {
244 1 : await connection.close();
245 1 : }
246 1 : } finally {
247 1 : await release();
248 1 : }
249 1 : },
250 1 : remove: async name => { await tryRemove(name); },
251 1 : };
252 1 : }
|
