pnpm

1
// cspell:ignore checkin
2
import path from 'path'
3
import os from 'os'
4
import { WorkerPool } from '@rushstack/worker-pool/lib/WorkerPool'
5
import { PnpmError } from '@pnpm/error'
6
import { type PackageFilesIndex } from '@pnpm/store.cafs'
7
import { type DependencyManifest } from '@pnpm/types'
8
import {
9
  type TarballExtractMessage,
10
  type AddDirToStoreMessage,
11
  type LinkPkgMessage,
12
  type SymlinkAllModulesMessage,
13
  type HardLinkDirMessage,
14
} from './types'
15

16
let workerPool: WorkerPool | undefined
17

18
export async function restartWorkerPool (): Promise<void> {
19
  await finishWorkers()
20
  workerPool = createTarballWorkerPool()
21
}
22

23
export async function finishWorkers (): Promise<void> {
24
  // @ts-expect-error
25
  await global.finishWorkers?.()
26
}
27

28
function createTarballWorkerPool (): WorkerPool {
29
  const maxWorkers = Math.max(2, (os.availableParallelism?.() ?? os.cpus().length) - Math.abs(process.env.PNPM_WORKERS ? parseInt(process.env.PNPM_WORKERS) : 0)) - 1
30
  const workerPool = new WorkerPool({
31
    id: 'pnpm',
32
    maxWorkers,
33
    workerScriptPath: path.join(__dirname, 'worker.js'),
34
  })
35
  // @ts-expect-error
36
  if (global.finishWorkers) {
37
    // @ts-expect-error
38
    const previous = global.finishWorkers
39
    // @ts-expect-error
40
    global.finishWorkers = async () => {
41
      await previous()
42
      await workerPool.finishAsync()
43
    }
44
  } else {
45
    // @ts-expect-error
46
    global.finishWorkers = () => workerPool.finishAsync()
47
  }
48
  return workerPool
49
}
50

51
interface AddFilesResult {
52
  filesIndex: Record<string, string>
53
  manifest: DependencyManifest
54
  requiresBuild: boolean
55
}
56

57
type AddFilesFromDirOptions = Pick<AddDirToStoreMessage, 'cafsDir' | 'dir' | 'filesIndexFile' | 'sideEffectsCacheKey' | 'readManifest' | 'pkg' | 'files'>
58

59
export async function addFilesFromDir (opts: AddFilesFromDirOptions): Promise<AddFilesResult> {
60
  if (!workerPool) {
61
    workerPool = createTarballWorkerPool()
62
  }
63
  const localWorker = await workerPool.checkoutWorkerAsync(true)
64
  return new Promise<{ filesIndex: Record<string, string>, manifest: DependencyManifest, requiresBuild: boolean }>((resolve, reject) => {
65
    localWorker.once('message', ({ status, error, value }) => {
66
      workerPool!.checkinWorker(localWorker)
67
      if (status === 'error') {
68
        reject(new PnpmError(error.code ?? 'GIT_FETCH_FAILED', error.message as string))
69
        return
70
      }
71
      resolve(value)
72
    })
73
    localWorker.postMessage({
74
      type: 'add-dir',
75
      cafsDir: opts.cafsDir,
76
      dir: opts.dir,
77
      filesIndexFile: opts.filesIndexFile,
78
      sideEffectsCacheKey: opts.sideEffectsCacheKey,
79
      readManifest: opts.readManifest,
80
      pkg: opts.pkg,
81
      files: opts.files,
82
    })
83
  })
84
}
85

86
export class TarballIntegrityError extends PnpmError {
87
  public readonly found: string
88
  public readonly expected: string
89
  public readonly algorithm: string
90
  public readonly sri: string
91
  public readonly url: string
92

93
  constructor (opts: {
94
    attempts?: number
95
    found: string
96
    expected: string
97
    algorithm: string
98
    sri: string
99
    url: string
100
  }) {
101
    super('TARBALL_INTEGRITY',
102
      `Got unexpected checksum for "${opts.url}". Wanted "${opts.expected}". Got "${opts.found}".`,
103
      {
104
        attempts: opts.attempts,
105
        hint: `This error may happen when a package is republished to the registry with the same version.
106
In this case, the metadata in the local pnpm cache will contain the old integrity checksum.
107

108
If you think that this is the case, then run "pnpm store prune" and rerun the command that failed.
109
"pnpm store prune" will remove your local metadata cache.`,
110
      }
111
    )
112
    this.found = opts.found
113
    this.expected = opts.expected
114
    this.algorithm = opts.algorithm
115
    this.sri = opts.sri
116
    this.url = opts.url
117
  }
118
}
119

120
type AddFilesFromTarballOptions = Pick<TarballExtractMessage, 'buffer' | 'cafsDir' | 'filesIndexFile' | 'integrity' | 'readManifest' | 'pkg'> & {
121
  url: string
122
}
123

124
export async function addFilesFromTarball (opts: AddFilesFromTarballOptions): Promise<AddFilesResult> {
125
  if (!workerPool) {
126
    workerPool = createTarballWorkerPool()
127
  }
128
  const localWorker = await workerPool.checkoutWorkerAsync(true)
129
  return new Promise<{ filesIndex: Record<string, string>, manifest: DependencyManifest, requiresBuild: boolean }>((resolve, reject) => {
130
    localWorker.once('message', ({ status, error, value }) => {
131
      workerPool!.checkinWorker(localWorker)
132
      if (status === 'error') {
133
        if (error.type === 'integrity_validation_failed') {
134
          reject(new TarballIntegrityError({
135
            ...error,
136
            url: opts.url,
137
          }))
138
          return
139
        }
140
        reject(new PnpmError(error.code ?? 'TARBALL_EXTRACT', `Failed to add tarball from "${opts.url}" to store: ${error.message as string}`))
141
        return
142
      }
143
      resolve(value)
144
    })
145
    localWorker.postMessage({
146
      type: 'extract',
147
      buffer: opts.buffer,
148
      cafsDir: opts.cafsDir,
149
      integrity: opts.integrity,
150
      filesIndexFile: opts.filesIndexFile,
151
      readManifest: opts.readManifest,
152
      pkg: opts.pkg,
153
    })
154
  })
155
}
156

157
export async function readPkgFromCafs (
158
  cafsDir: string,
159
  verifyStoreIntegrity: boolean,
160
  filesIndexFile: string,
161
  readManifest?: boolean
162
): Promise<{ verified: boolean, pkgFilesIndex: PackageFilesIndex, manifest?: DependencyManifest, requiresBuild: boolean }> {
163
  if (!workerPool) {
164
    workerPool = createTarballWorkerPool()
165
  }
166
  const localWorker = await workerPool.checkoutWorkerAsync(true)
167
  return new Promise<{ verified: boolean, pkgFilesIndex: PackageFilesIndex, requiresBuild: boolean }>((resolve, reject) => {
168
    localWorker.once('message', ({ status, error, value }) => {
169
      workerPool!.checkinWorker(localWorker)
170
      if (status === 'error') {
171
        reject(new PnpmError(error.code ?? 'READ_FROM_STORE', error.message as string))
172
        return
173
      }
174
      resolve(value)
175
    })
176
    localWorker.postMessage({
177
      type: 'readPkgFromCafs',
178
      cafsDir,
179
      filesIndexFile,
180
      readManifest,
181
      verifyStoreIntegrity,
182
    })
183
  })
184
}
185

186
export async function importPackage (
187
  opts: Omit<LinkPkgMessage, 'type'>
188
): Promise<{ isBuilt: boolean, importMethod: string | undefined }> {
189
  if (!workerPool) {
190
    workerPool = createTarballWorkerPool()
191
  }
192
  const localWorker = await workerPool.checkoutWorkerAsync(true)
193
  return new Promise<{ isBuilt: boolean, importMethod: string | undefined }>((resolve, reject) => {
194
    localWorker.once('message', ({ status, error, value }: any) => { // eslint-disable-line @typescript-eslint/no-explicit-any
195
      workerPool!.checkinWorker(localWorker)
196
      if (status === 'error') {
197
        reject(new PnpmError(error.code ?? 'LINKING_FAILED', error.message as string))
198
        return
199
      }
200
      resolve(value)
201
    })
202
    localWorker.postMessage({
203
      type: 'link',
204
      ...opts,
205
    })
206
  })
207
}
208

209
export async function symlinkAllModules (
210
  opts: Omit<SymlinkAllModulesMessage, 'type'>
211
): Promise<{ isBuilt: boolean, importMethod: string | undefined }> {
212
  if (!workerPool) {
213
    workerPool = createTarballWorkerPool()
214
  }
215
  const localWorker = await workerPool.checkoutWorkerAsync(true)
216
  return new Promise<{ isBuilt: boolean, importMethod: string | undefined }>((resolve, reject) => {
217
    localWorker.once('message', ({ status, error, value }: any) => { // eslint-disable-line @typescript-eslint/no-explicit-any
218
      workerPool!.checkinWorker(localWorker)
219
      if (status === 'error') {
220
        reject(new PnpmError(error.code ?? 'SYMLINK_FAILED', error.message as string))
221
        return
222
      }
223
      resolve(value)
224
    })
225
    localWorker.postMessage({
226
      type: 'symlinkAllModules',
227
      ...opts,
228
    } as SymlinkAllModulesMessage)
229
  })
230
}
231

232
export async function hardLinkDir (src: string, destDirs: string[]): Promise<void> {
233
  if (!workerPool) {
234
    workerPool = createTarballWorkerPool()
235
  }
236
  const localWorker = await workerPool.checkoutWorkerAsync(true)
237
  await new Promise<void>((resolve, reject) => {
238
    localWorker.once('message', ({ status, error }: any) => { // eslint-disable-line @typescript-eslint/no-explicit-any
239
      workerPool!.checkinWorker(localWorker)
240
      if (status === 'error') {
241
        reject(new PnpmError(error.code ?? 'HARDLINK_FAILED', error.message as string))
242
        return
243
      }
244
      resolve()
245
    })
246
    localWorker.postMessage({
247
      type: 'hardLinkDir',
248
      src,
249
      destDirs,
250
    } as HardLinkDirMessage)
251
  })
252
}
253