|
@@ -0,0 +1,564 @@
|
|
|
|
|
+// @ts-check
|
|
|
|
|
+import logger from '@overleaf/logger'
|
|
|
|
|
+import zlib from 'node:zlib'
|
|
|
|
|
+import Settings from '@overleaf/settings'
|
|
|
|
|
+import Path from 'node:path'
|
|
|
|
|
+import fs from 'node:fs'
|
|
|
|
|
+import CLSICacheHandler from './CLSICacheHandler.js'
|
|
|
|
|
+import Errors from './Errors.js'
|
|
|
|
|
+import { callbackify, promisify } from 'node:util'
|
|
|
|
|
+import {
|
|
|
|
|
+ AddFileOperation,
|
|
|
|
|
+ Change,
|
|
|
|
|
+ EditFileOperation,
|
|
|
|
|
+ File,
|
|
|
|
|
+ MoveFileOperation,
|
|
|
|
|
+ Snapshot,
|
|
|
|
|
+} from 'overleaf-editor-core'
|
|
|
|
|
+import { fetchString, RequestFailedError } from '@overleaf/fetch-utils'
|
|
|
|
|
+import { setTimeout } from 'node:timers/promises'
|
|
|
|
|
+import ResourceWriter from './ResourceWriter.js'
|
|
|
|
|
+import UrlCache from './UrlCache.js'
|
|
|
|
|
+import OError from '@overleaf/o-error'
|
|
|
|
|
+import ClsiMetrics from './Metrics.js'
|
|
|
|
|
+import { promiseMapSettledWithLimit } from '@overleaf/promise-utils'
|
|
|
|
|
+
|
|
|
|
|
+const gzip = promisify(zlib.gzip)
|
|
|
|
|
+const gunzip = promisify(zlib.gunzip)
|
|
|
|
|
+
|
|
|
|
|
+export const clearCacheCb = callbackify(clearCache)
|
|
|
|
|
+
|
|
|
|
|
+/**
|
|
|
|
|
+ * @param {string} projectId
|
|
|
|
|
+ * @param {string} userId
|
|
|
|
|
+ * @return {Promise<void>}
|
|
|
|
|
+ */
|
|
|
|
|
+export async function clearCache(projectId, userId) {
|
|
|
|
|
+ const { dir } = snapshotPath(projectId, userId)
|
|
|
|
|
+ try {
|
|
|
|
|
+ await fs.promises.rm(dir, { recursive: true, force: true })
|
|
|
|
|
+ } catch (err) {
|
|
|
|
|
+ if (isENOENT(err)) return
|
|
|
|
|
+ logger.warn(
|
|
|
|
|
+ { err, projectId, userId },
|
|
|
|
|
+ 'compile from cache: failed to clear history cache'
|
|
|
|
|
+ )
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+/**
|
|
|
|
|
+ * @param {string} projectId
|
|
|
|
|
+ * @param {string} userId
|
|
|
|
|
+ * @return {{ dir: string, path: string, resyncPath: string }}
|
|
|
|
|
+ */
|
|
|
|
|
+function snapshotPath(projectId, userId) {
|
|
|
|
|
+ const dir = Path.join(
|
|
|
|
|
+ Settings.path.clsiCacheDir,
|
|
|
|
|
+ userId ? `${projectId}-${userId}` : projectId
|
|
|
|
|
+ )
|
|
|
|
|
+
|
|
|
|
|
+ const path = Path.join(dir, 'history.json.gz')
|
|
|
|
|
+ const resyncPath = Path.join(dir, 'history-resync.json.gz')
|
|
|
|
|
+ return { dir, path, resyncPath }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+/**
|
|
|
|
|
+ * @param {unknown} err
|
|
|
|
|
+ * @return {boolean}
|
|
|
|
|
+ */
|
|
|
|
|
+function isENOENT(err) {
|
|
|
|
|
+ return err instanceof Error && 'code' in err && err.code === 'ENOENT'
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+/**
|
|
|
|
|
+ * @param {string} projectId
|
|
|
|
|
+ * @param {string} userId
|
|
|
|
|
+ * @param {number} remoteBaseVersion
|
|
|
|
|
+ * @return {Promise<{rawSnapshot: import('overleaf-editor-core/lib/types.js').RawSnapshot, globalBlobs: string[], fullSync: boolean,localBaseVersion: number}>}
|
|
|
|
|
+ */
|
|
|
|
|
+async function loadSnapshot(projectId, userId, remoteBaseVersion) {
|
|
|
|
|
+ const { path, resyncPath } = snapshotPath(projectId, userId)
|
|
|
|
|
+ let maxLocalBaseVersion = -1
|
|
|
|
|
+ for (const candidate of [path, resyncPath]) {
|
|
|
|
|
+ try {
|
|
|
|
|
+ const fullSync = candidate === resyncPath
|
|
|
|
|
+ return await loadSnapshotFromFile(candidate, remoteBaseVersion, fullSync)
|
|
|
|
|
+ } catch (err) {
|
|
|
|
|
+ if (err instanceof Errors.MissingUpdatesError) {
|
|
|
|
|
+ maxLocalBaseVersion = Math.max(
|
|
|
|
|
+ maxLocalBaseVersion,
|
|
|
|
|
+ err.info.baseHistoryVersion
|
|
|
|
|
+ )
|
|
|
|
|
+ } else if (!isENOENT(err)) {
|
|
|
|
|
+ logger.warn(
|
|
|
|
|
+ { err, projectId, userId },
|
|
|
|
|
+ 'compile from cache: cannot read history from disk'
|
|
|
|
|
+ )
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ try {
|
|
|
|
|
+ return await loadSnapshotFromClsiCache(projectId, userId, remoteBaseVersion)
|
|
|
|
|
+ } catch (err) {
|
|
|
|
|
+ if (err instanceof Errors.MissingUpdatesError) {
|
|
|
|
|
+ maxLocalBaseVersion = Math.max(
|
|
|
|
|
+ maxLocalBaseVersion,
|
|
|
|
|
+ err.info.baseHistoryVersion
|
|
|
|
|
+ )
|
|
|
|
|
+ } else if (!isENOENT(err)) {
|
|
|
|
|
+ logger.warn(
|
|
|
|
|
+ { err, projectId, userId },
|
|
|
|
|
+ 'compile from cache: cannot download from clsi-cache'
|
|
|
|
|
+ )
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ throw new Errors.MissingUpdatesError('needs more updates', {
|
|
|
|
|
+ baseHistoryVersion: maxLocalBaseVersion,
|
|
|
|
|
+ })
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+/**
|
|
|
|
|
+ * @param {string} projectId
|
|
|
|
|
+ * @param {string} userId
|
|
|
|
|
+ * @param {number} remoteBaseVersion
|
|
|
|
|
+ * @return {Promise<{rawSnapshot: import('overleaf-editor-core/lib/types.js').RawSnapshot, globalBlobs: string[], fullSync: boolean,localBaseVersion: number}>}
|
|
|
|
|
+ */
|
|
|
|
|
+async function loadSnapshotFromClsiCache(projectId, userId, remoteBaseVersion) {
|
|
|
|
|
+ const { dir, resyncPath } = snapshotPath(projectId, userId)
|
|
|
|
|
+ await fs.promises.mkdir(dir, { recursive: true })
|
|
|
|
|
+ const ok = await CLSICacheHandler.downloadHistorySnapshot(
|
|
|
|
|
+ projectId,
|
|
|
|
|
+ userId,
|
|
|
|
|
+ dir
|
|
|
|
|
+ )
|
|
|
|
|
+ if (!ok) {
|
|
|
|
|
+ throw new Errors.MissingUpdatesError('needs full sync', {
|
|
|
|
|
+ baseHistoryVersion: -1,
|
|
|
|
|
+ })
|
|
|
|
|
+ }
|
|
|
|
|
+ logger.debug(
|
|
|
|
|
+ { projectId, userId },
|
|
|
|
|
+ 'compile from cache: restored history from clsi-cache'
|
|
|
|
|
+ )
|
|
|
|
|
+ return await loadSnapshotFromFile(resyncPath, remoteBaseVersion, true)
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+/**
|
|
|
|
|
+ * @param {string} path
|
|
|
|
|
+ * @param {number} remoteBaseVersion
|
|
|
|
|
+ * @param {boolean} fullSync
|
|
|
|
|
+ * @return {Promise<{rawSnapshot: import('overleaf-editor-core/lib/types.js').RawSnapshot, globalBlobs: string[], localBaseVersion: number, fullSync: boolean}>}
|
|
|
|
|
+ */
|
|
|
|
|
+async function loadSnapshotFromFile(path, remoteBaseVersion, fullSync) {
|
|
|
|
|
+ let blob = await fs.promises.readFile(path)
|
|
|
|
|
+ blob = await gunzip(blob)
|
|
|
|
|
+ const { rawSnapshot, globalBlobs, localBaseVersion } = JSON.parse(
|
|
|
|
|
+ blob.toString('utf-8')
|
|
|
|
|
+ )
|
|
|
|
|
+ if (localBaseVersion < remoteBaseVersion) {
|
|
|
|
|
+ throw new Errors.MissingUpdatesError('missing updates', {
|
|
|
|
|
+ baseHistoryVersion: localBaseVersion,
|
|
|
|
|
+ })
|
|
|
|
|
+ }
|
|
|
|
|
+ return { rawSnapshot, globalBlobs, localBaseVersion, fullSync }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+/**
|
|
|
|
|
+ * @param {string} projectId
|
|
|
|
|
+ * @param {string} userId
|
|
|
|
|
+ * @param {Snapshot} snapshot
|
|
|
|
|
+ * @param {number} localBaseVersion
|
|
|
|
|
+ * @param {string[]} globalBlobs
|
|
|
|
|
+ * @return {Promise<void>}
|
|
|
|
|
+ */
|
|
|
|
|
+async function saveSnapshot(
|
|
|
|
|
+ projectId,
|
|
|
|
|
+ userId,
|
|
|
|
|
+ snapshot,
|
|
|
|
|
+ localBaseVersion,
|
|
|
|
|
+ globalBlobs
|
|
|
|
|
+) {
|
|
|
|
|
+ const { dir, path } = snapshotPath(projectId, userId)
|
|
|
|
|
+ await fs.promises.mkdir(dir, { recursive: true })
|
|
|
|
|
+ const tmp = path + '~'
|
|
|
|
|
+ await fs.promises.writeFile(
|
|
|
|
|
+ tmp,
|
|
|
|
|
+ await gzip(
|
|
|
|
|
+ JSON.stringify({
|
|
|
|
|
+ globalBlobs,
|
|
|
|
|
+ localBaseVersion,
|
|
|
|
|
+ rawSnapshot: snapshot.toRaw(),
|
|
|
|
|
+ })
|
|
|
|
|
+ ),
|
|
|
|
|
+ { flag: 'wx' }
|
|
|
|
|
+ )
|
|
|
|
|
+ await fs.promises.rename(tmp, path)
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+/**
|
|
|
|
|
+ * @param {string} projectId
|
|
|
|
|
+ * @param {string} userId
|
|
|
|
|
+ * @return {Promise<void>}
|
|
|
|
|
+ */
|
|
|
|
|
+async function deleteResyncSnapshot(projectId, userId) {
|
|
|
|
|
+ const { resyncPath } = snapshotPath(projectId, userId)
|
|
|
|
|
+ try {
|
|
|
|
|
+ await fs.promises.unlink(resyncPath)
|
|
|
|
|
+ } catch (err) {
|
|
|
|
|
+ if (!isENOENT(err)) {
|
|
|
|
|
+ logger.warn(
|
|
|
|
|
+ { err, projectId, userId },
|
|
|
|
|
+ 'compile from cache: failed to clear history-resync.json.gz'
|
|
|
|
|
+ )
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+/**
|
|
|
|
|
+ * @param {string} compileDir
|
|
|
|
|
+ * @param {string} subDir
|
|
|
|
|
+ * @param {Map<string, boolean>} entries
|
|
|
|
|
+ * @return {Promise<Map<string, boolean>>}
|
|
|
|
|
+ */
|
|
|
|
|
+async function discoverExistingEntries(
|
|
|
|
|
+ compileDir,
|
|
|
|
|
+ subDir = '.',
|
|
|
|
|
+ entries = new Map()
|
|
|
|
|
+) {
|
|
|
|
|
+ const dirents = await fs.promises.readdir(Path.join(compileDir, subDir), {
|
|
|
|
|
+ withFileTypes: true,
|
|
|
|
|
+ })
|
|
|
|
|
+ for (const dirent of dirents) {
|
|
|
|
|
+ const path = Path.join(subDir, dirent.name)
|
|
|
|
|
+ if (dirent.isDirectory()) {
|
|
|
|
|
+ await discoverExistingEntries(compileDir, path, entries)
|
|
|
|
|
+ } else if (dirent.isFile()) {
|
|
|
|
|
+ entries.set(path, false)
|
|
|
|
|
+ } else if (
|
|
|
|
|
+ dirent.isSymbolicLink() ||
|
|
|
|
|
+ dirent.isFIFO() ||
|
|
|
|
|
+ dirent.isSocket()
|
|
|
|
|
+ ) {
|
|
|
|
|
+ // should not happen, delete right away
|
|
|
|
|
+ logger.warn(
|
|
|
|
|
+ { compileDir, subDir, dirent },
|
|
|
|
|
+ 'compile from cache: found blocked dirent'
|
|
|
|
|
+ )
|
|
|
|
|
+ await fs.promises.unlink(Path.join(compileDir, path))
|
|
|
|
|
+ } else {
|
|
|
|
|
+ throw new OError('unexpected dir entry', { compileDir, subDir, dirent })
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ entries.set(subDir, true)
|
|
|
|
|
+ return entries
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+/**
|
|
|
|
|
+ * @param {string} compileDir
|
|
|
|
|
+ * @param {Snapshot} snapshot
|
|
|
|
|
+ * @param {Map<string, boolean>} entriesDepthFirst
|
|
|
|
|
+ */
|
|
|
|
|
+async function removeExtraneousEntries(
|
|
|
|
|
+ compileDir,
|
|
|
|
|
+ snapshot,
|
|
|
|
|
+ entriesDepthFirst
|
|
|
|
|
+) {
|
|
|
|
|
+ const keepFolders = new Set(['.'])
|
|
|
|
|
+ for (const [path, isDir] of entriesDepthFirst) {
|
|
|
|
|
+ const shouldBeFile = !!snapshot.getFile(path)
|
|
|
|
|
+ if (isDir) {
|
|
|
|
|
+ if (!shouldBeFile) {
|
|
|
|
|
+ // directory can stay directory
|
|
|
|
|
+ if (keepFolders.has(path)) {
|
|
|
|
|
+ // folder is still in use
|
|
|
|
|
+ keepFolders.add(Path.dirname(path))
|
|
|
|
|
+ } else {
|
|
|
|
|
+ // empty folder
|
|
|
|
|
+ await fs.promises.rmdir(Path.join(compileDir, path))
|
|
|
|
|
+ entriesDepthFirst.delete(path)
|
|
|
|
|
+ }
|
|
|
|
|
+ continue
|
|
|
|
|
+ }
|
|
|
|
|
+ // a folder turned into a file
|
|
|
|
|
+ // before: foo/bar.txt/baz.txt
|
|
|
|
|
+ // ^^^^^^^ folder
|
|
|
|
|
+ // now: foo/bar.txt
|
|
|
|
|
+ // ^^^^^^^ file
|
|
|
|
|
+ const needle = path + '/'
|
|
|
|
|
+ for (const [child, childIsDir] of entriesDepthFirst) {
|
|
|
|
|
+ if (!child.startsWith(needle)) continue
|
|
|
|
|
+ if (childIsDir) {
|
|
|
|
|
+ await fs.promises.rmdir(Path.join(compileDir, child))
|
|
|
|
|
+ } else {
|
|
|
|
|
+ await fs.promises.unlink(Path.join(compileDir, child))
|
|
|
|
|
+ }
|
|
|
|
|
+ entriesDepthFirst.delete(child)
|
|
|
|
|
+ }
|
|
|
|
|
+ await fs.promises.rmdir(Path.join(compileDir, path))
|
|
|
|
|
+ entriesDepthFirst.delete(path)
|
|
|
|
|
+ continue
|
|
|
|
|
+ }
|
|
|
|
|
+ if (shouldBeFile || !ResourceWriter.isExtraneousFile(path)) {
|
|
|
|
|
+ // resource or cached file
|
|
|
|
|
+ keepFolders.add(Path.dirname(path))
|
|
|
|
|
+ continue
|
|
|
|
|
+ }
|
|
|
|
|
+ await fs.promises.unlink(Path.join(compileDir, path))
|
|
|
|
|
+ entriesDepthFirst.delete(path)
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+/**
|
|
|
|
|
+ * @param {string} compileDir
|
|
|
|
|
+ * @param {string} path
|
|
|
|
|
+ * @param {Map<string, boolean>} entriesDepthFirst
|
|
|
|
|
+ */
|
|
|
|
|
+async function ensureHasParentFolder(compileDir, path, entriesDepthFirst) {
|
|
|
|
|
+ const parentFolderPath = Path.dirname(path)
|
|
|
|
|
+ if (entriesDepthFirst.has(parentFolderPath)) return
|
|
|
|
|
+ await ensureHasParentFolder(compileDir, parentFolderPath, entriesDepthFirst)
|
|
|
|
|
+ await fs.promises.mkdir(Path.join(compileDir, parentFolderPath))
|
|
|
|
|
+ entriesDepthFirst.set(parentFolderPath, true)
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+/**
|
|
|
|
|
+ * @param {import('overleaf-editor-core/lib/types.js').RawOperation[][]} raw
|
|
|
|
|
+ * @return {Change[]}
|
|
|
|
|
+ */
|
|
|
|
|
+function changesFromRawChangeOperations(raw) {
|
|
|
|
|
+ return raw.map(o => Change.mustFromRaw({ operations: o, timestamp: '0' }))
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+/**
|
|
|
|
|
+ * @param {string} projectId
|
|
|
|
|
+ * @param {string} userId
|
|
|
|
|
+ * @param {Object} request
|
|
|
|
|
+ * @param {string} compileDir
|
|
|
|
|
+ * @param {Record<string, number>} timings
|
|
|
|
|
+ * @return {Promise<{baseHistoryVersion: number, resourceList: {path: string}[]}>}
|
|
|
|
|
+ */
|
|
|
|
|
+export async function syncResourcesToDisk(
|
|
|
|
|
+ projectId,
|
|
|
|
|
+ userId,
|
|
|
|
|
+ request,
|
|
|
|
|
+ compileDir,
|
|
|
|
|
+ timings
|
|
|
|
|
+) {
|
|
|
|
|
+ const remoteBaseVersion = request.baseHistoryVersion
|
|
|
|
|
+ let rawSnapshot, globalBlobs, localBaseVersion, source
|
|
|
|
|
+ let fullSync = true
|
|
|
|
|
+ try {
|
|
|
|
|
+ ;({ rawSnapshot, globalBlobs, fullSync, localBaseVersion } =
|
|
|
|
|
+ await loadSnapshot(projectId, userId, remoteBaseVersion))
|
|
|
|
|
+ source = fullSync ? 'clsi-cache' : 'local'
|
|
|
|
|
+ logger.debug(
|
|
|
|
|
+ { projectId, userId, localBaseVersion, remoteBaseVersion },
|
|
|
|
|
+ 'compile from cache: using existing snapshot'
|
|
|
|
|
+ )
|
|
|
|
|
+ } catch (err) {
|
|
|
|
|
+ if (!request.rawSnapshot) throw err
|
|
|
|
|
+ if (!(err instanceof Errors.MissingUpdatesError)) {
|
|
|
|
|
+ logger.warn(
|
|
|
|
|
+ { err, projectId, userId },
|
|
|
|
|
+ 'compile from cache: bad local history state during full resync'
|
|
|
|
|
+ )
|
|
|
|
|
+ }
|
|
|
|
|
+ logger.debug(
|
|
|
|
|
+ { projectId, userId },
|
|
|
|
|
+ 'compile from cache: using incoming snapshot'
|
|
|
|
|
+ )
|
|
|
|
|
+ source = 'remote'
|
|
|
|
|
+ localBaseVersion = remoteBaseVersion
|
|
|
|
|
+ rawSnapshot = request.rawSnapshot
|
|
|
|
|
+ globalBlobs = []
|
|
|
|
|
+ }
|
|
|
|
|
+ globalBlobs = Array.from(new Set(globalBlobs.concat(request.globalBlobs)))
|
|
|
|
|
+
|
|
|
|
|
+ const snapshot = Snapshot.fromRaw(rawSnapshot)
|
|
|
|
|
+
|
|
|
|
|
+ const changes = changesFromRawChangeOperations(
|
|
|
|
|
+ request.rawChangeOperations.slice(localBaseVersion - remoteBaseVersion)
|
|
|
|
|
+ )
|
|
|
|
|
+ const applyAllStart = performance.now()
|
|
|
|
|
+ snapshot.applyAll(changes)
|
|
|
|
|
+ timings.snapshotApplyAll = Math.ceil(performance.now() - applyAllStart)
|
|
|
|
|
+ if (!ClsiMetrics.shouldSkipMetrics(request)) {
|
|
|
|
|
+ ClsiMetrics.snapshotApplyAllDurationSeconds.observe(
|
|
|
|
|
+ { group: request.compileGroup, source },
|
|
|
|
|
+ timings.snapshotApplyAll / 1_000
|
|
|
|
|
+ )
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ const entriesDepthFirst = await discoverExistingEntries(compileDir)
|
|
|
|
|
+ await removeExtraneousEntries(compileDir, snapshot, entriesDepthFirst)
|
|
|
|
|
+
|
|
|
|
|
+ const changedPaths = []
|
|
|
|
|
+ if (fullSync) {
|
|
|
|
|
+ changedPaths.push(...snapshot.getFilePathnames())
|
|
|
|
|
+ logger.debug({ projectId, userId }, 'compile from cache: full sync')
|
|
|
|
|
+ } else {
|
|
|
|
|
+ const dedupe = new Set()
|
|
|
|
|
+ for (const change of changes) {
|
|
|
|
|
+ for (const operation of change.getOperations()) {
|
|
|
|
|
+ if (operation instanceof AddFileOperation) {
|
|
|
|
|
+ dedupe.add(operation.pathname)
|
|
|
|
|
+ } else if (operation instanceof MoveFileOperation) {
|
|
|
|
|
+ dedupe.add(operation.pathname)
|
|
|
|
|
+ if (!operation.isRemoveFile()) dedupe.add(operation.newPathname)
|
|
|
|
|
+ } else if (operation instanceof EditFileOperation) {
|
|
|
|
|
+ dedupe.add(operation.pathname)
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ // Restore deleted files
|
|
|
|
|
+ for (const path of snapshot.getFilePathnames()) {
|
|
|
|
|
+ if (!entriesDepthFirst.has(path)) dedupe.add(path)
|
|
|
|
|
+ }
|
|
|
|
|
+ changedPaths.push(...dedupe)
|
|
|
|
|
+ logger.debug(
|
|
|
|
|
+ { projectId, userId, changedPaths },
|
|
|
|
|
+ 'compile from cache: incremental sync'
|
|
|
|
|
+ )
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ const blobStore = new BlobStore(request.historyId, globalBlobs)
|
|
|
|
|
+ const loadEagerStart = performance.now()
|
|
|
|
|
+ await snapshot.loadFiles('eager', blobStore)
|
|
|
|
|
+ timings.snapshotLoadEager = Math.ceil(performance.now() - loadEagerStart)
|
|
|
|
|
+ if (!ClsiMetrics.shouldSkipMetrics(request)) {
|
|
|
|
|
+ ClsiMetrics.snapshotLoadEagerDurationSeconds.observe(
|
|
|
|
|
+ { group: request.compileGroup, source },
|
|
|
|
|
+ timings.snapshotLoadEager / 1_000
|
|
|
|
|
+ )
|
|
|
|
|
+ }
|
|
|
|
|
+ for (const path of changedPaths) {
|
|
|
|
|
+ const file = snapshot.getFile(path)
|
|
|
|
|
+ if (!file) continue // deleted, handled by removeExtraneousEntries
|
|
|
|
|
+ await ensureHasParentFolder(compileDir, path, entriesDepthFirst)
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ let createCacheFolder
|
|
|
|
|
+ // Use Promise.allSettled to ensure that all writes have stopped when we exit.
|
|
|
|
|
+ const allDone = await promiseMapSettledWithLimit(
|
|
|
|
|
+ Settings.parallelFileDownloads,
|
|
|
|
|
+ changedPaths,
|
|
|
|
|
+ async path => {
|
|
|
|
|
+ const file = snapshot.getFile(path)
|
|
|
|
|
+ if (!file) return // deleted, handled by removeExtraneousEntries
|
|
|
|
|
+
|
|
|
|
|
+ const content = file.getContent({ filterTrackedDeletes: true })
|
|
|
|
|
+ if (typeof content === 'string') {
|
|
|
|
|
+ await fs.promises.writeFile(
|
|
|
|
|
+ Path.join(compileDir, path),
|
|
|
|
|
+ content,
|
|
|
|
|
+ 'utf-8'
|
|
|
|
|
+ )
|
|
|
|
|
+ } else {
|
|
|
|
|
+ const hash = file.getHash()
|
|
|
|
|
+ if (!hash) {
|
|
|
|
|
+ throw new OError('unexpected file without content and hash', { path })
|
|
|
|
|
+ }
|
|
|
|
|
+ const fallbackURL = null // no fallback
|
|
|
|
|
+ const lastModified = new Date(0) // content is static
|
|
|
|
|
+ if (!createCacheFolder) {
|
|
|
|
|
+ createCacheFolder = UrlCache.promises.createProjectDir(projectId)
|
|
|
|
|
+ }
|
|
|
|
|
+ await createCacheFolder
|
|
|
|
|
+ await UrlCache.promises.downloadUrlToFile(
|
|
|
|
|
+ projectId,
|
|
|
|
|
+ blobStore.getBlobURL(hash).href,
|
|
|
|
|
+ fallbackURL,
|
|
|
|
|
+ Path.join(compileDir, path),
|
|
|
|
|
+ lastModified
|
|
|
|
|
+ )
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ )
|
|
|
|
|
+ for (const [idx, result] of allDone.entries()) {
|
|
|
|
|
+ if (result.status === 'fulfilled') continue
|
|
|
|
|
+ const path = changedPaths[idx]
|
|
|
|
|
+ throw OError.tag(result.reason, 'write failed', { path })
|
|
|
|
|
+ }
|
|
|
|
|
+ const baseHistoryVersion = localBaseVersion + changes.length
|
|
|
|
|
+ if (fullSync || changes.length) {
|
|
|
|
|
+ await saveSnapshot(
|
|
|
|
|
+ projectId,
|
|
|
|
|
+ userId,
|
|
|
|
|
+ snapshot,
|
|
|
|
|
+ baseHistoryVersion,
|
|
|
|
|
+ globalBlobs
|
|
|
|
|
+ )
|
|
|
|
|
+ }
|
|
|
|
|
+ if (fullSync) {
|
|
|
|
|
+ await deleteResyncSnapshot(projectId, userId)
|
|
|
|
|
+ }
|
|
|
|
|
+ return {
|
|
|
|
|
+ baseHistoryVersion,
|
|
|
|
|
+ resourceList: snapshot.getFilePathnames().map(path => ({ path })),
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+class BlobStore {
|
|
|
|
|
+ /** @type {string} */
|
|
|
|
|
+ #historyId
|
|
|
|
|
+ /** @type {string[]} */
|
|
|
|
|
+ #globalBlobs
|
|
|
|
|
+
|
|
|
|
|
+ /**
|
|
|
|
|
+ * @param {string} historyId
|
|
|
|
|
+ * @param {string[]} globalBlobs
|
|
|
|
|
+ */
|
|
|
|
|
+ constructor(historyId, globalBlobs) {
|
|
|
|
|
+ this.#historyId = historyId
|
|
|
|
|
+ this.#globalBlobs = globalBlobs
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ /**
|
|
|
|
|
+ * @param {string} hash
|
|
|
|
|
+ * @return {URL}
|
|
|
|
|
+ */
|
|
|
|
|
+ getBlobURL(hash) {
|
|
|
|
|
+ const u = new URL(Settings.apis.filestore.url)
|
|
|
|
|
+ if (this.#globalBlobs.includes(hash)) {
|
|
|
|
|
+ u.pathname = `/history/global/hash/${hash}`
|
|
|
|
|
+ } else {
|
|
|
|
|
+ u.pathname = `/history/project/${this.#historyId}/hash/${hash}`
|
|
|
|
|
+ }
|
|
|
|
|
+ return u
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ /**
|
|
|
|
|
+ * @param {string} hash
|
|
|
|
|
+ * @return {Promise<string>}
|
|
|
|
|
+ */
|
|
|
|
|
+ async getString(hash) {
|
|
|
|
|
+ if (hash === File.EMPTY_FILE_HASH) return ''
|
|
|
|
|
+ const u = this.getBlobURL(hash)
|
|
|
|
|
+ let remainingAttempts = 3
|
|
|
|
|
+ while (true) {
|
|
|
|
|
+ try {
|
|
|
|
|
+ return await fetchString(u, { signal: AbortSignal.timeout(3_000) })
|
|
|
|
|
+ } catch (err) {
|
|
|
|
|
+ if (err instanceof RequestFailedError && err.response.status === 404) {
|
|
|
|
|
+ throw new Errors.NotFoundError()
|
|
|
|
|
+ }
|
|
|
|
|
+ remainingAttempts--
|
|
|
|
|
+ if (remainingAttempts <= 0) throw err
|
|
|
|
|
+ logger.warn(
|
|
|
|
|
+ { err, url: u.href, remainingAttempts },
|
|
|
|
|
+ 'compile from cache: history blob download failed'
|
|
|
|
|
+ )
|
|
|
|
|
+ await setTimeout(100)
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ /**
|
|
|
|
|
+ * @param {string} hash
|
|
|
|
|
+ * @return {Promise<any>}
|
|
|
|
|
+ */
|
|
|
|
|
+ async getObject(hash) {
|
|
|
|
|
+ const string = await this.getString(hash)
|
|
|
|
|
+ return JSON.parse(string)
|
|
|
|
|
+ }
|
|
|
|
|
+}
|