diff --git a/apps/heft/src/index.ts b/apps/heft/src/index.ts index 7ccd112e96..e51062cddb 100644 --- a/apps/heft/src/index.ts +++ b/apps/heft/src/index.ts @@ -79,6 +79,8 @@ export { export type { IScopedLogger } from './pluginFramework/logging/ScopedLogger'; +export type { IShardResultManifest, ShardStatus } from './sharding/IShardResultManifest'; + // Re-export types required to use custom command-line parameters export type { CommandLineParameter, diff --git a/apps/heft/src/schemas/shard-result-manifest.schema.json b/apps/heft/src/schemas/shard-result-manifest.schema.json new file mode 100644 index 0000000000..bccce62c85 --- /dev/null +++ b/apps/heft/src/schemas/shard-result-manifest.schema.json @@ -0,0 +1,110 @@ +{ + "$schema": "http://json-schema.org/draft-04/schema#", + "title": "Heft Shard Result Manifest", + "description": "This schema describes the \"shard-manifest.json\" file that each shard writes into its output directory to report what it did.", + "type": "object", + + "additionalProperties": false, + + "required": [ + "version", + "shardIndex", + "shardCount", + "generation", + "producerAttempt", + "status", + "workload", + "startedAt", + "durationMs", + "diagnosticCount", + "itemCount" + ], + + "properties": { + "version": { + "title": "Manifest Version", + "description": "The version of this manifest format. A collator must refuse a version it does not understand rather than guessing at the shape.", + "enum": [1] + }, + + "shardIndex": { + "title": "Shard Index", + "description": "The 1-based index of the shard that produced this manifest, matching the numbering used by both Rush and Jest.", + "type": "integer", + "minimum": 1, + "maximum": 9007199254740991 + }, + + "shardCount": { + "title": "Shard Count", + "description": "The total number of shards the producing shard believed it was part of. The collator compares this against its own shard count to detect a reconfiguration that a cache hit would otherwise hide.", + "type": "integer", + "minimum": 1, + "maximum": 9007199254740991 + }, + + "generation": { + "title": "Shard Generation", + "description": "An opaque token identifying the sharded run that produced this manifest. The collator requires it to be uniform across every manifest, which is what prevents results from two different runs being merged together.", + "type": "string", + "minLength": 1 + }, + + "producerAttempt": { + "title": "Producer Attempt", + "description": "A non-semantic execution-attempt token used for live publication fencing and provenance. It is not part of semantic cache identity.", + "type": "string", + "minLength": 1 + }, + + "status": { + "title": "Shard Status", + "description": "\"completed\" means the shard ran its workload to completion, whether or not that workload reported failures. \"harness-failure\" means the shard could not run its workload at all, so its counts carry no information.", + "enum": ["completed", "harness-failure"] + }, + + "workload": { + "title": "Workload Name", + "description": "The kind of work this shard performed, for example \"jest\" or \"lint\". The neutral core treats this as opaque; only an adapter interprets it.", + "type": "string", + "minLength": 1 + }, + + "startedAt": { + "title": "Started At", + "description": "An ISO 8601 timestamp recording when the shard began its workload.", + "type": "string", + "format": "date-time", + "minLength": 1 + }, + + "durationMs": { + "title": "Duration In Milliseconds", + "description": "How long the shard's workload took, in milliseconds.", + "type": "number", + "minimum": 0, + "maximum": 9007199254740991 + }, + + "diagnosticCount": { + "title": "Diagnostic Count", + "description": "The number of problems the shard found, for example failing tests or lint errors. The collator sums this across shards to reach its verdict.", + "type": "integer", + "minimum": 0, + "maximum": 9007199254740991 + }, + + "itemCount": { + "title": "Item Count", + "description": "The number of units of work the shard actually processed, for example tests run or files linted. The adapter applies its whole-workload empty-selection policy.", + "type": "integer", + "minimum": 0, + "maximum": 9007199254740991 + }, + + "payload": { + "title": "Adapter Payload", + "description": "Adapter-specific data that the neutral core passes through without inspecting. A Jest adapter uses this to carry the information its merge step needs." + } + } +} diff --git a/apps/heft/src/sharding/IShardResultManifest.ts b/apps/heft/src/sharding/IShardResultManifest.ts new file mode 100644 index 0000000000..e4db5f212b --- /dev/null +++ b/apps/heft/src/sharding/IShardResultManifest.ts @@ -0,0 +1,111 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +/** + * Whether a shard was able to run its workload at all. + * + * @remarks + * This is deliberately separate from whether the workload found problems. A shard that runs + * every test it was given and observes ten failures is `completed`; a shard whose test runner + * crashed before it could start is `harness-failure`. A consumer must not interpret a + * `harness-failure` shard's `diagnosticCount` and `itemCount` as completed work. + * + * @public + */ +export type ShardStatus = 'completed' | 'harness-failure'; + +/** + * What a single shard reports about the work it performed. + * + * @remarks + * Producers write this as `shard-manifest.json` in their owned output directory after + * publishing all payload files. A successful exit without a manifest is not completion evidence. + * + * The fields above `payload` are deliberately workload-agnostic: `diagnosticCount` and + * `itemCount` are generic enough that validation and verdict logic need no knowledge of what + * the shard was actually doing. Anything workload-specific belongs in `payload`. + * + * @public + */ +export interface IShardResultManifest { + /** + * The version of this manifest format. + */ + version: 1; + + /** + * The 1-based index of the shard that produced this manifest, matching the numbering used by + * both Rush and Jest. + */ + shardIndex: number; + + /** + * The total number of shards the producing shard believed it was part of. + * + * @remarks + * Consumers must compare this against the expected shard count. Without it, changing the shard count + * and then hitting the build cache for a stale shard would silently produce a partial result. + */ + shardCount: number; + + /** + * An opaque token identifying the sharded run that produced this manifest. + * + * @remarks + * Consumers must compare this with the current invocation's independently supplied generation, + * not just with sibling manifests. A uniformly stale set is not a current result. + */ + generation: string; + + /** + * A non-semantic token identifying the live execution attempt that published this manifest. + * + * @remarks + * Unlike {@link IShardResultManifest.generation}, this changes for equivalent retries and does + * not participate in cache identity. It records which fenced writer performed the final commit. + */ + producerAttempt: string; + + /** + * Whether the shard ran its workload to completion. + */ + status: ShardStatus; + + /** + * The kind of work this shard performed, for example `jest` or `lint`. + */ + workload: string; + + /** + * An ISO 8601 timestamp recording when the shard began its workload. + */ + startedAt: string; + + /** + * How long the shard's workload took, in milliseconds. + */ + durationMs: number; + + /** + * The number of problems the shard found, for example failing tests or lint errors. + */ + diagnosticCount: number; + + /** + * The number of units of work the shard actually processed, for example tests run or files + * linted. + * + * @remarks + * The workload consumer is responsible for its empty-workload policy. + */ + itemCount: number; + + /** + * Adapter-specific data that the neutral core passes through without inspecting. + * + * @remarks + * The generic schema does not interpret this field. Workload consumers must validate their + * own payload format before using it. Producers with no additional data may omit it. + */ + payload?: unknown; +} diff --git a/apps/heft/src/sharding/ShardFileSystem.ts b/apps/heft/src/sharding/ShardFileSystem.ts new file mode 100644 index 0000000000..9b9374c57e --- /dev/null +++ b/apps/heft/src/sharding/ShardFileSystem.ts @@ -0,0 +1,141 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import fs from 'node:fs/promises'; +import fsSync, { type BigIntStats } from 'node:fs'; +import path from 'node:path'; +import { randomUUID } from 'node:crypto'; + +import { FileSystem } from '@rushstack/node-core-library'; + +/** + * Identifies an atomic publication so cancellation cannot invalidate a replacement file. + * Callers must retain exclusive destination ownership through their final cancellation check. + * @public + */ +export interface IShardFilePublication { + invalidateIfOwned(): void; +} + +/** + * An exclusive live claim that serializes revocation and final promotion. + * The callback must execute synchronously and cannot release the claim. + * @public + */ +export interface IShardFilePublicationGuard { + commit(action: () => void): void; +} + +/** + * Rejects paths outside the build folder and symbolic links/junctions below it. + * The build folder itself may be a symlinked workspace. + * @public + */ +export async function validateShardPathAsync( + buildFolderPath: string, + targetPath: string, + abortSignal?: AbortSignal +): Promise { + const relative: string = path.relative(buildFolderPath, targetPath); + if (!relative || relative === '..' || relative.startsWith(`..${path.sep}`) || path.isAbsolute(relative)) { + throw new Error(`Shard paths must be below the build folder: "${targetPath}".`); + } + let current: string = buildFolderPath; + for (const segment of relative.split(path.sep)) { + abortSignal?.throwIfAborted(); + current = path.join(current, segment); + try { + if ((await fs.lstat(current)).isSymbolicLink()) { + throw new Error(`Shard paths must not traverse a symbolic link: "${current}".`); + } + } catch (error) { + if (!FileSystem.isNotExistError(error as Error)) throw error; + } + abortSignal?.throwIfAborted(); + } +} + +/** + * Removes only a known output file, tolerating absence but not other filesystem errors. + * @public + */ +export async function invalidateShardFileAsync(filename: string): Promise { + try { + await fs.unlink(filename); + } catch (error) { + if (!FileSystem.isNotExistError(error as Error)) throw error; + } +} + +/** + * Publishes a complete file by same-directory atomic rename. Callers must invalidate their + * previous output before starting work, and validate the destination before calling this helper. + * @public + */ +export async function writeShardFileAtomicallyAsync( + filename: string, + contents: string, + abortSignal?: AbortSignal, + publicationGuard?: IShardFilePublicationGuard +): Promise { + abortSignal?.throwIfAborted(); + const folder: string = path.dirname(filename); + await fs.mkdir(folder, { recursive: true }); + abortSignal?.throwIfAborted(); + const temporaryPath: string = path.join(folder, `.${path.basename(filename)}.${randomUUID()}.tmp`); + const handle: fs.FileHandle = await fs.open(temporaryPath, 'wx'); + let published: boolean = false; + let identity: BigIntStats; + try { + try { + abortSignal?.throwIfAborted(); + identity = await handle.stat({ bigint: true }); + await handle.writeFile(contents); + await handle.sync(); + abortSignal?.throwIfAborted(); + } finally { + await handle.close(); + } + abortSignal?.throwIfAborted(); + const publication: IShardFilePublication = { + invalidateIfOwned: () => { + let current: BigIntStats; + try { + current = fsSync.lstatSync(filename, { bigint: true }); + } catch (error) { + if (!FileSystem.isNotExistError(error as Error)) throw error; + return; + } + if ( + current.isFile() && + current.ino === identity.ino && + // Windows lstat reports dev=0; fstat reports the volume serial number. + (process.platform === 'win32' || current.dev === identity.dev) + ) { + fsSync.unlinkSync(filename); + } + } + }; + if (publicationGuard) { + // No await between the ownership check and rename: revocation/release in this process + // cannot interleave, and other processes cannot acquire the exclusively held claim. + publicationGuard.commit(() => { + abortSignal?.throwIfAborted(); + fsSync.renameSync(temporaryPath, filename); + published = true; + }); + } else { + await fs.rename(temporaryPath, filename); + published = true; + } + if (abortSignal?.aborted) { + publication.invalidateIfOwned(); + abortSignal.throwIfAborted(); + } + return publication; + } finally { + // Rename consumes the staging file. No asynchronous cleanup may follow the final + // cancellation check on the success path. + if (!published) await FileSystem.deleteFileAsync(temporaryPath); + } +} diff --git a/apps/heft/src/sharding/ShardIdentity.ts b/apps/heft/src/sharding/ShardIdentity.ts new file mode 100644 index 0000000000..c7fae8c3ed --- /dev/null +++ b/apps/heft/src/sharding/ShardIdentity.ts @@ -0,0 +1,118 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import path from 'node:path'; + +/** + * Which shard of a sharded run the current process is. + * + * @public + */ +export interface IShardIdentity { + /** + * The 1-based index of this shard. + * + * @remarks + * Rush's shard loop is 1-based, and every mainstream test runner's `--shard` syntax is + * 1-based as well, so no conversion is performed anywhere in this package. + */ + shardIndex: number; + + /** + * The total number of shards in this run. + */ + shardCount: number; +} + +/** + * The `k/n` syntax shared by Rush's default `shardArgumentFormat` and every mainstream + * test runner's `--shard` parameter. + * + * @remarks + * The pattern deliberately anchors both ends and allows no whitespace, sign, or separator + * beyond a single `/`. Jest's own parser splits on `/` and then *filters out* the + * non-numeric segments before checking that two remain, which means it silently accepts + * inputs such as `"x/1/2"` as shard 1 of 2. Reproducing that behavior would turn a typo + * into a run that quietly tests a fraction of the suite, so this parser rejects it. + */ +const SHARD_PAIR_REGEXP: RegExp = /^(\d+)\/(\d+)$/; + +/** + * Parses the `k/n` value of a `--shard` parameter. + * + * @remarks + * The returned value is validated to the same rules the underlying runners enforce, but it + * is validated *here* so that the failure is reported against the Heft parameter the user + * actually typed, before any workload starts. + * + * @param value - the raw parameter value, for example `"2/4"` + * @returns the parsed shard identity + * + * @public + */ +export function parseShardPair(value: string): IShardIdentity { + const match: RegExpExecArray | null = SHARD_PAIR_REGEXP.exec(value); + if (!match || match[0] !== value) { + throw new Error( + `Invalid "--shard" value ${JSON.stringify(value)}: expected the format "/",` + + ` for example "2/4".` + ); + } + + const shardIndex: number = Number.parseInt(match[1], 10); + const shardCount: number = Number.parseInt(match[2], 10); + + if (!Number.isSafeInteger(shardIndex) || !Number.isSafeInteger(shardCount)) { + throw new Error( + `Invalid "--shard" value ${JSON.stringify(value)}: index and count must be safe integers.` + ); + } + + if (shardIndex < 1 || shardCount < 1) { + throw new Error( + `Invalid "--shard" value ${JSON.stringify(value)}: the index and count are 1-based, so neither` + + ` may be zero.` + ); + } + + if (shardIndex > shardCount) { + throw new Error( + `Invalid "--shard" value ${JSON.stringify(value)}: the index ${shardIndex} is greater than the` + + ` count ${shardCount}.` + ); + } + + return { shardIndex, shardCount }; +} + +/** + * Formats a shard identity back into the `k/n` syntax. + * + * @public + */ +export function formatShardPair(identity: IShardIdentity): string { + return `${identity.shardIndex}/${identity.shardCount}`; +} + +/** + * Returns the name of the shard-scoped subfolder for the given shard. + * + * @remarks + * Concurrent shards of the same operation run in the same project folder, so any state a + * workload keeps under its temp folder — a Jest haste map, a cache, a lockfile — is shared + * by default. Every shard-scoped path in this package is derived from this one function so + * that the isolation cannot be applied inconsistently. + * + * @public + */ +export function getShardFolderName(shardIndex: number): string { + return `shard-${shardIndex}`; +} + +/** + * Returns an isolated cache folder, including the partition count to avoid reusing a different partition. + * @public + */ +export function getShardTempFolderPath(tempFolderPath: string, identity: IShardIdentity): string { + return path.join(tempFolderPath, `${getShardFolderName(identity.shardIndex)}-of-${identity.shardCount}`); +} diff --git a/apps/heft/src/sharding/ShardOutputOwnership.ts b/apps/heft/src/sharding/ShardOutputOwnership.ts new file mode 100644 index 0000000000..dce4126838 --- /dev/null +++ b/apps/heft/src/sharding/ShardOutputOwnership.ts @@ -0,0 +1,520 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import fs from 'node:fs/promises'; +import fsSync, { type BigIntStats } from 'node:fs'; +import path from 'node:path'; +import { createHash, randomUUID } from 'node:crypto'; +import { hostname } from 'node:os'; +import { setTimeout as sleepAsync } from 'node:timers/promises'; + +import { FileSystem } from '@rushstack/node-core-library'; + +import { validateShardPathAsync } from './ShardFileSystem'; + +// Keep the protocol namespace stable across workload adapters so they contend on the same claims. +export const SHARD_OUTPUT_OWNERSHIP_FOLDER_NAME: '.heft-jest-shard-locks' = '.heft-jest-shard-locks'; + +export type ShardOutputClaimState = 'held' | 'publishing' | 'revoked' | 'released'; + +export interface IShardOutputOwner { + project?: string; + phase?: string; + task?: string; + shardIndex?: number; + shardCount?: number; + generation?: string; +} + +export interface IShardOutputLock { + readonly filePath: string; + readonly attemptId: string; + readonly state: ShardOutputClaimState; + readonly isReleased: boolean; + assertOwned(): void; + /** Synchronously fence and promote a staged file without an await between check and rename. */ + commit(action: () => void): void; + /** Revocation prevents new publication but retains the claim until all work and cleanup settle. */ + revoke(): void; + release(): void; +} + +interface IOutputClaim extends IShardOutputOwner { + version: 2; + kind: 'registration' | 'output'; + destination: string; + attemptId: string; + host: string; + pid: number; + processStartedAt: string; + acquiredAt: string; +} + +const PROCESS_STARTED_AT: string = new Date(Date.now() - process.uptime() * 1000).toISOString(); +const RECOVERY_GUIDANCE: string = + 'Use distinct output directories for distinct operations. For an abandoned claim, remove the ownership file only ' + + 'after confirming all processes of its owning operation have exited and pending writes have settled. ' + + 'Never steal a claim based only on PID or age; never clean active ownership files or include them in shard caches.'; + +/** Resolve existing ancestors without materializing missing input directories or result files. */ +export async function resolveShardPathAsync(filename: string): Promise { + try { + return await fs.realpath(filename); + } catch (error) { + if (!FileSystem.isNotExistError(error as Error)) throw error; + return path.join(await resolveShardPathAsync(path.dirname(filename)), path.basename(filename)); + } +} + +export function getShardOutputIdentity(buildFolderPath: string, outputPath: string): string { + const relative: string = path.relative(buildFolderPath, outputPath); + const identity: string = relative.normalize('NFC'); + if (!identity || identity === '..' || identity.startsWith(`..${path.sep}`) || path.isAbsolute(identity)) { + throw new Error(`Managed shard output must be below the build folder: "${outputPath}".`); + } + const reservedIdentity: string = identity.toLowerCase(); + if ( + reservedIdentity === SHARD_OUTPUT_OWNERSHIP_FOLDER_NAME || + reservedIdentity.startsWith(`${SHARD_OUTPUT_OWNERSHIP_FOLDER_NAME}${path.sep}`) + ) { + throw new Error( + `Managed shard output must not use the reserved "${SHARD_OUTPUT_OWNERSHIP_FOLDER_NAME}" directory.` + ); + } + return identity; +} + +async function doesShardProbeExistAsync(filename: string): Promise { + try { + await fs.lstat(filename); + return true; + } catch (error) { + if (FileSystem.isNotExistError(error as Error)) return false; + throw error; + } +} + +async function areFileNamesEquivalentAsync( + probeParentPath: string, + first: string, + second: string +): Promise { + if (first === second) return true; + // Keep complete components unchanged, including names at the filesystem's component-length limit. + const probePath: string = await fs.mkdtemp(path.join(probeParentPath, '.heft-shard-name-probe-')); + try { + const parentIgnoresCase: boolean = await doesShardProbeExistAsync( + path.join(probeParentPath, path.basename(probePath).toUpperCase()) + ); + const caseProbePath: string = path.join(probePath, 'case-probe'); + await fs.mkdir(caseProbePath); + const probeIgnoresCase: boolean = await doesShardProbeExistAsync(path.join(probePath, 'CASE-PROBE')); + await fs.rmdir(caseProbePath); + // Per-directory case settings are not universally inherited. Never fall back to the lock directory. + if (parentIgnoresCase !== probeIgnoresCase) { + throw new Error( + `Cannot compare managed shard output names: the probe directory does not preserve the case ` + + `sensitivity of "${probeParentPath}".` + ); + } + await fs.mkdir(path.join(probePath, first)); + return await doesShardProbeExistAsync(path.join(probePath, second)); + } finally { + await fs.rm(probePath, { recursive: true, force: true }); + } +} + +async function doShardOutputIdentitiesOverlapAsync( + buildFolderPath: string, + firstOutputPath: string, + second: string +): Promise { + const firstSegments: string[] = path.relative(buildFolderPath, firstOutputPath).split(path.sep); + const secondSegments: string[] = second.split(path.sep); + const sharedLength: number = Math.min(firstSegments.length, secondSegments.length); + let parentPath: string = buildFolderPath; + for (let index: number = 0; index < sharedLength; index++) { + const first: string = firstSegments[index].normalize('NFC'); + if (first !== secondSegments[index]) { + if (parentPath !== buildFolderPath) await validateShardPathAsync(buildFolderPath, parentPath); + if (!(await areFileNamesEquivalentAsync(parentPath, first, secondSegments[index]))) return false; + } + // Keep the physical spelling of existing ancestors, not their normalized claim identity. + parentPath = path.join(parentPath, firstSegments[index]); + } + return true; +} + +function gateContentionError(filePath: string, retryCount: number, deadline: number): Error { + return new Error( + `Timed out waiting to register managed shard output ownership through "${filePath}". ` + + `state=retry-exhausted retryCount=${retryCount} retryDeadline=${new Date(deadline).toISOString()}. ` + + describeHolder(filePath) + + ' ' + + RECOVERY_GUIDANCE + ); +} + +function isOutputClaim(value: unknown): value is IOutputClaim { + if (!value || typeof value !== 'object') return false; + const claim: Partial = value; + return ( + claim.version === 2 && + (claim.kind === 'registration' || claim.kind === 'output') && + typeof claim.destination === 'string' && + typeof claim.attemptId === 'string' && + claim.attemptId.length > 0 && + typeof claim.host === 'string' && + typeof claim.pid === 'number' && + Number.isSafeInteger(claim.pid) && + claim.pid > 0 && + typeof claim.processStartedAt === 'string' && + Number.isFinite(Date.parse(claim.processStartedAt)) && + typeof claim.acquiredAt === 'string' && + Number.isFinite(Date.parse(claim.acquiredAt)) && + [claim.project, claim.phase, claim.task, claim.generation].every( + (field) => field === undefined || typeof field === 'string' + ) && + [claim.shardIndex, claim.shardCount].every( + (field) => + field === undefined || (typeof field === 'number' && Number.isSafeInteger(field) && field > 0) + ) + ); +} + +function describeHolder(filePath: string): string { + try { + const stats: BigIntStats = fsSync.lstatSync(filePath, { bigint: true }); + if (!stats.isFile() || stats.nlink !== 1n) return 'Holder metadata is not a singly linked regular file.'; + const value: unknown = JSON.parse(fsSync.readFileSync(filePath, 'utf8')); + if (!isOutputClaim(value)) + return 'Holder metadata is malformed or from an older runtime; do not reclaim automatically.'; + return `holder=${JSON.stringify({ + project: value.project, + phase: value.phase, + task: value.task, + shardIndex: value.shardIndex, + shardCount: value.shardCount, + generation: value.generation, + attemptId: value.attemptId, + destination: value.destination, + host: value.host, + pid: value.pid, + processStartedAt: value.processStartedAt, + acquiredAt: value.acquiredAt, + claimAgeMs: Math.max(0, Date.now() - Date.parse(value.acquiredAt)) + })}`; + } catch (error) { + return `Holder metadata could not be read: ${String(error)}.`; + } +} + +/** + * Materialize directories, never result files, before deriving ownership. Promise realpath + * uses native Windows semantics, including 8.3 aliases; the JS realpathSync implementation does not. + */ +export async function resolveShardOutputPathAsync( + buildFolderPath: string, + outputPath: string, + directory: boolean +): Promise { + getShardOutputIdentity(buildFolderPath, outputPath); + await validateShardPathAsync(buildFolderPath, outputPath); + const realBuild: string = await fs.realpath(buildFolderPath); + const target: string = path.resolve(realBuild, path.relative(buildFolderPath, outputPath)); + const parent: string = directory ? target : path.dirname(target); + await fs.mkdir(parent, { recursive: true }); + const realParent: string = await fs.realpath(parent); + let canonical: string = realParent; + if (!directory) { + canonical = path.join(realParent, path.basename(target)); + try { + canonical = await fs.realpath(canonical); + } catch (error) { + if (!FileSystem.isNotExistError(error as Error)) throw error; + } + } + await validateShardPathAsync(realBuild, canonical); + getShardOutputIdentity(realBuild, canonical); + return canonical; +} + +function ownershipError(filePath: string): Error { + return new Error( + `Managed shard output is already owned by another task or process, or an interrupted owner left "${filePath}". ` + + 'state=contended retryCount=0 (destination conflicts fail immediately). ' + + describeHolder(filePath) + + ' ' + + RECOVERY_GUIDANCE + ); +} + +/** No PID inference or stale-lock deletion: unlinking an open file can steal a live Windows owner's output. */ +function acquireOutputLock( + filePath: string, + claim: IOutputClaim, + reportContention: boolean = true, + abortSignal?: AbortSignal +): IShardOutputLock { + let descriptor: number; + try { + descriptor = fsSync.openSync(filePath, 'wx'); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'EEXIST' || !reportContention) throw error; + throw ownershipError(filePath); + } + let identity: BigIntStats; + try { + identity = fsSync.fstatSync(descriptor, { bigint: true }); + } catch (error) { + const errors: unknown[] = [error]; + try { + fsSync.unlinkSync(filePath); + } catch (unlinkError) { + if (!FileSystem.isNotExistError(unlinkError as Error)) errors.push(unlinkError); + } + try { + fsSync.closeSync(descriptor); + } catch (closeError) { + errors.push(closeError); + } + if (errors.length === 1) throw error; + throw new AggregateError( + errors, + `Failed to inspect and clean up managed shard ownership: "${filePath}".` + ); + } + let released: boolean = false; + let revoked: boolean = false; + let publishing: boolean = false; + const revoke = (): void => { + revoked = true; + }; + const assertIdentity = (): void => { + if (released) throw new Error(`Managed shard ownership was already released: "${filePath}".`); + const current: BigIntStats = fsSync.lstatSync(filePath, { bigint: true }); + if ( + !current.isFile() || + current.nlink !== 1n || + // Windows lstat reports dev=0, whereas fstat reports the volume serial number. + (process.platform !== 'win32' && current.dev !== identity.dev) || + current.ino !== identity.ino + ) { + throw new Error(`Managed shard ownership file changed while in use: "${filePath}".`); + } + }; + const assertOwned = (): void => { + assertIdentity(); + if (revoked) throw new Error(`Shard attempt "${claim.attemptId}" was revoked: "${filePath}".`); + }; + const lock: IShardOutputLock = { + filePath, + attemptId: claim.attemptId, + get state(): ShardOutputClaimState { + return released ? 'released' : revoked ? 'revoked' : publishing ? 'publishing' : 'held'; + }, + get isReleased(): boolean { + return released; + }, + assertOwned, + revoke, + commit: (action) => { + assertOwned(); + if (publishing) throw new Error(`Shard attempt is already publishing: "${filePath}".`); + publishing = true; + try { + action(); + } finally { + publishing = false; + } + }, + release: () => { + if (released) throw new Error(`Managed shard ownership was already released: "${filePath}".`); + if (publishing) throw new Error(`Cannot release a publishing shard claim: "${filePath}".`); + revoke(); + try { + assertIdentity(); + } finally { + fsSync.closeSync(descriptor); + released = true; + abortSignal?.removeEventListener('abort', revoke); + } + fsSync.unlinkSync(filePath); + } + }; + try { + fsSync.writeFileSync(filePath, JSON.stringify(claim)); + } catch (error) { + try { + lock.release(); + } catch (releaseError) { + throw new AggregateError( + [error, releaseError], + `Failed to write and release shard claim "${filePath}".` + ); + } + throw error; + } + abortSignal?.addEventListener('abort', revoke, { once: true }); + if (abortSignal?.aborted) revoke(); + return lock; +} + +export function releaseShardOutputLocks(locks: Iterable): void { + const errors: unknown[] = []; + for (const lock of locks) { + try { + lock.release(); + } catch (error) { + errors.push(error); + } + } + if (errors.length) throw new AggregateError(errors, 'Failed to release managed shard output ownership.'); +} + +/** + * Claim canonical destinations, including their descendants. A short registration gate makes + * overlap checking and multi-output acquisition atomic across processes. It is never held during + * workload execution. An abandoned gate, like an abandoned output claim, requires manual recovery. + * Call resolveShardOutputPathAsync() first for every destination; this function does not resolve aliases. + * Differing names require an existing, writable parent for temporary name-equivalence probes. + * Callers must also keep outputs separate from source files and other inputs. + */ +export async function acquireShardOutputLocksAsync( + buildFolderPath: string, + outputPaths: Iterable, + owner: IShardOutputOwner = {}, + abortSignal?: AbortSignal +): Promise { + const realBuild: string = await fs.realpath(buildFolderPath); + const lockFolder: string = path.join(realBuild, SHARD_OUTPUT_OWNERSHIP_FOLDER_NAME); + await validateShardPathAsync(realBuild, lockFolder); + await fs.mkdir(lockFolder, { recursive: true }); + const destinations: string[] = []; + const destinationPaths: Map = new Map(); + for (const output of outputPaths) { + const destination: string = getShardOutputIdentity(realBuild, output); + if (!destinationPaths.has(destination)) { + destinationPaths.set(destination, output); + destinations.push(destination); + } + } + for (let currentIndex: number = 0; currentIndex < destinations.length; currentIndex++) { + for (let previousIndex: number = 0; previousIndex < currentIndex; previousIndex++) { + if ( + await doShardOutputIdentitiesOverlapAsync( + realBuild, + destinationPaths.get(destinations[previousIndex])!, + destinations[currentIndex] + ) + ) { + throw new Error( + `Managed shard outputs overlap within one operation: "${destinations[previousIndex]}" and ` + + `"${destinations[currentIndex]}".` + ); + } + } + } + const gatePath: string = path.join(lockFolder, 'acquire.lock'); + await validateShardPathAsync(realBuild, gatePath); + const attemptId: string = randomUUID(); + const makeClaim = (destination: string, kind: IOutputClaim['kind']): IOutputClaim => ({ + ...owner, + version: 2, + kind, + destination, + attemptId, + host: hostname(), + pid: process.pid, + processStartedAt: PROCESS_STARTED_AT, + acquiredAt: new Date().toISOString() + }); + let gate: IShardOutputLock | undefined; + const gateDeadline: number = Date.now() + 2000; + let retryCount: number = 0; + while (!gate) { + try { + gate = acquireOutputLock(gatePath, makeClaim(destinations.join(', '), 'registration'), false); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'EEXIST') throw error; + if (Date.now() >= gateDeadline) throw gateContentionError(gatePath, retryCount, gateDeadline); + retryCount++; + await sleepAsync(5 + Math.floor(Math.random() * 16)); + } + } + const locks: IShardOutputLock[] = []; + let acquisitionError: unknown; + try { + for (const entry of fsSync.readdirSync(lockFolder, { withFileTypes: true })) { + if (entry.name === 'acquire.lock' || !entry.name.endsWith('.lock')) continue; + const filename: string = path.join(lockFolder, entry.name); + let claim: IOutputClaim; + try { + const stats: BigIntStats = fsSync.lstatSync(filename, { bigint: true }); + if (!stats.isFile() || stats.nlink !== 1n) throw ownershipError(filename); + const value: unknown = JSON.parse(fsSync.readFileSync(filename, 'utf8')); + if ( + !isOutputClaim(value) || + value.kind !== 'output' || + getShardOutputIdentity(realBuild, path.resolve(realBuild, value.destination)) !== + value.destination || + `${createHash('sha256').update(value.destination).digest('hex')}.lock` !== entry.name + ) { + throw ownershipError(filename); + } + claim = value; + } catch (error) { + // A normal release can race directory enumeration, but a malformed/legacy claim is not reclaimable. + if (FileSystem.isNotExistError(error as Error)) continue; + throw ownershipError(filename); + } + for (const outputPath of destinationPaths.values()) { + if (await doShardOutputIdentitiesOverlapAsync(realBuild, outputPath, claim.destination)) { + throw ownershipError(filename); + } + } + } + for (const destination of destinations) { + const filename: string = path.join( + lockFolder, + `${createHash('sha256').update(destination).digest('hex')}.lock` + ); + const lock: IShardOutputLock = acquireOutputLock( + filename, + makeClaim(destination, 'output'), + true, + abortSignal + ); + locks.push(lock); + } + } catch (error) { + acquisitionError = error; + } + if (acquisitionError !== undefined) { + const errors: unknown[] = [acquisitionError]; + try { + releaseShardOutputLocks(locks); + } catch (error) { + errors.push(error); + } + try { + gate.release(); + } catch (error) { + errors.push(error); + } + if (errors.length === 1) throw acquisitionError; + throw new AggregateError(errors, 'Failed to register and clean up managed shard output ownership.'); + } + try { + gate.release(); + } catch (error) { + const errors: unknown[] = [error]; + try { + releaseShardOutputLocks(locks); + } catch (releaseError) { + errors.push(releaseError); + } + throw new AggregateError(errors, 'Failed to finalize managed shard output ownership registration.'); + } + return locks; +} diff --git a/apps/heft/src/sharding/ShardResultManifest.ts b/apps/heft/src/sharding/ShardResultManifest.ts new file mode 100644 index 0000000000..d246f2a074 --- /dev/null +++ b/apps/heft/src/sharding/ShardResultManifest.ts @@ -0,0 +1,126 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import path from 'node:path'; + +import { FileSystem, JsonFile, JsonSchema } from '@rushstack/node-core-library'; + +import schema from '../schemas/shard-result-manifest.schema.json'; +import type { IShardResultManifest } from './IShardResultManifest'; +import { + invalidateShardFileAsync, + writeShardFileAtomicallyAsync, + type IShardFilePublication, + type IShardFilePublicationGuard +} from './ShardFileSystem'; + +/** + * The name of the file each shard writes into its output directory to report what it did. + * + * @public + */ +export const SHARD_MANIFEST_FILENAME: 'shard-manifest.json' = 'shard-manifest.json'; + +const MANIFEST_SCHEMA: JsonSchema = JsonSchema.fromLoadedObject(schema); + +/** + * Validates an already-parsed manifest without reading files. + * @public + */ +export function validateShardResultManifest( + value: unknown, + filename: string = 'Shard result manifest' +): IShardResultManifest { + MANIFEST_SCHEMA.validateObject(value, filename); + return value as IShardResultManifest; +} + +/** + * Returns the path of the manifest file within the specified shard output folder. + * + * @public + */ +export function getShardManifestPath(shardFolderPath: string): string { + return path.join(shardFolderPath, SHARD_MANIFEST_FILENAME); +} + +/** + * Invalidates only this shard's commit marker before starting a new attempt. + * A crash or configuration error must not leave an earlier success visible. + * + * @public + */ +export async function invalidateShardResultManifestAsync(shardFolderPath: string): Promise { + await invalidateShardFileAsync(getShardManifestPath(shardFolderPath)); +} + +/** + * Writes a shard's manifest into its output folder. + * + * @remarks + * A shard must call this on every path where it exits successfully, including when its workload + * reported failures. The collator cannot distinguish a shard that ran and left nothing behind + * from a shard that never ran at all. + * + * @public + */ +export async function writeShardResultManifestAsync( + shardFolderPath: string, + manifest: IShardResultManifest, + abortSignal?: AbortSignal, + publicationGuard?: IShardFilePublicationGuard +): Promise { + abortSignal?.throwIfAborted(); + const manifestPath: string = getShardManifestPath(shardFolderPath); + const serialized: string = JSON.stringify(manifest); + validateShardResultManifest(JSON.parse(serialized), manifestPath); + // The manifest is a commit marker: publish it only after every payload/report is complete. + const publication: IShardFilePublication = await writeShardFileAtomicallyAsync( + manifestPath, + serialized, + abortSignal, + publicationGuard + ); + if (abortSignal?.aborted) { + publication.invalidateIfOwned(); + abortSignal.throwIfAborted(); + } +} + +/** + * Reads and validates the manifest in the specified shard output folder. + * + * @remarks + * Throws if the manifest is absent, unparseable, or does not satisfy the manifest schema. + * + * @public + */ +export async function readShardResultManifestAsync(shardFolderPath: string): Promise { + const manifestPath: string = getShardManifestPath(shardFolderPath); + return (await JsonFile.loadAndValidateAsync(manifestPath, MANIFEST_SCHEMA)) as IShardResultManifest; +} + +/** + * Reads and validates the manifest in the specified shard output folder, returning `undefined` + * if no manifest is present. + * + * @remarks + * Only absence produces `undefined`. A manifest that exists but is unparseable or invalid still + * throws, because a corrupt result must never be quietly downgraded to a missing one -- the + * collator reacts to those two cases differently. + * + * @public + */ +export async function tryReadShardResultManifestAsync( + shardFolderPath: string +): Promise { + try { + return await readShardResultManifestAsync(shardFolderPath); + } catch (error) { + if (FileSystem.isNotExistError(error as Error)) { + return undefined; + } + + throw error; + } +} diff --git a/apps/heft/src/sharding/test/ShardFileSystem.test.ts b/apps/heft/src/sharding/test/ShardFileSystem.test.ts new file mode 100644 index 0000000000..a3ecb894a9 --- /dev/null +++ b/apps/heft/src/sharding/test/ShardFileSystem.test.ts @@ -0,0 +1,101 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import fs from 'node:fs/promises'; +import path from 'node:path'; + +import { + invalidateShardFileAsync, + validateShardPathAsync, + writeShardFileAtomicallyAsync +} from '../ShardFileSystem'; + +describe('atomic shard files', () => { + let folder: string; + beforeEach(async () => { + folder = await fs.mkdtemp(path.join(__dirname, 'atomic-shard-')); + }); + afterEach(async () => { + jest.restoreAllMocks(); + await fs.rm(folder, { recursive: true, force: true }); + }); + + it('replaces a report atomically without leaving staging files', async () => { + const filename: string = path.join(folder, 'report.json'); + await fs.writeFile(filename, 'old'); + await invalidateShardFileAsync(filename); + await writeShardFileAtomicallyAsync(filename, '{"success":false}'); + expect(await fs.readFile(filename, 'utf8')).toBe('{"success":false}'); + expect(await fs.readdir(folder)).toEqual(['report.json']); + }); + + it('cleans its staging file when rename fails, without deleting the destination', async () => { + const filename: string = path.join(folder, 'report.json'); + await fs.mkdir(filename); + await fs.writeFile(path.join(filename, 'sentinel'), 'preserve'); + await expect(writeShardFileAtomicallyAsync(filename, '{}')).rejects.toThrow(); + expect(await fs.readdir(folder)).toEqual(['report.json']); + expect(await fs.readFile(path.join(filename, 'sentinel'), 'utf8')).toBe('preserve'); + }); + + it('propagates cancellation without creating directories or reports', async () => { + const controller: AbortController = new AbortController(); + controller.abort(new Error('cancelled write')); + await expect( + writeShardFileAtomicallyAsync(path.join(folder, 'nested', 'report.json'), '{}', controller.signal) + ).rejects.toThrow('cancelled write'); + expect(await fs.readdir(folder)).toEqual([]); + }); + + it('cleans staging files when the publication guard rejects a write', async () => { + const filename: string = path.join(folder, 'report.json'); + await fs.writeFile(filename, 'previous'); + await expect( + writeShardFileAtomicallyAsync(filename, 'replacement', undefined, { + commit: () => { + throw new Error('ownership revoked'); + } + }) + ).rejects.toThrow('ownership revoked'); + expect(await fs.readFile(filename, 'utf8')).toBe('previous'); + expect(await fs.readdir(folder)).toEqual(['report.json']); + }); + + it('rejects the build root itself and validates missing descendants without creating them', async () => { + await expect(validateShardPathAsync(folder, folder)).rejects.toThrow(/below the build folder/); + await validateShardPathAsync(folder, path.join(folder, 'missing', 'report.json')); + expect(await fs.readdir(folder)).toEqual([]); + }); + + it('allows a symlinked workspace root but rejects links below that root', async () => { + const root: string = path.join(folder, 'root'); + const alias: string = path.join(folder, 'alias'); + await fs.mkdir(root); + await fs.symlink(root, alias, 'junction'); + await validateShardPathAsync(alias, path.join(alias, 'report.json')); + await fs.symlink(root, path.join(root, 'nested-alias'), 'junction'); + await expect( + validateShardPathAsync(alias, path.join(alias, 'nested-alias', 'report.json')) + ).rejects.toThrow(/symbolic link/); + }); + + it('propagates filesystem errors other than absence', async () => { + jest.spyOn(fs, 'unlink').mockRejectedValue(Object.assign(new Error('access denied'), { code: 'EACCES' })); + await expect(invalidateShardFileAsync(path.join(folder, 'report.json'))).rejects.toThrow('access denied'); + jest.spyOn(fs, 'lstat').mockRejectedValue(Object.assign(new Error('cannot inspect'), { code: 'EACCES' })); + await expect(validateShardPathAsync(folder, path.join(folder, 'report.json'))).rejects.toThrow( + 'cannot inspect' + ); + }); + + it('rejects a path that aliases a sibling via a junction', async () => { + await fs.mkdir(path.join(folder, 'actual')); + await fs.symlink(path.join(folder, 'actual'), path.join(folder, 'alias'), 'junction'); + await expect(validateShardPathAsync(folder, path.join(folder, 'alias', 'report.json'))).rejects.toThrow( + /symbolic link/ + ); + await expect(validateShardPathAsync(folder, path.resolve(folder, '..', 'outside'))).rejects.toThrow( + /below the build folder/ + ); + }); +}); diff --git a/apps/heft/src/sharding/test/ShardIdentity.test.ts b/apps/heft/src/sharding/test/ShardIdentity.test.ts new file mode 100644 index 0000000000..5b0b421bc5 --- /dev/null +++ b/apps/heft/src/sharding/test/ShardIdentity.test.ts @@ -0,0 +1,105 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import path from 'node:path'; + +import { + formatShardPair, + getShardFolderName, + getShardTempFolderPath, + parseShardPair +} from '../ShardIdentity'; + +describe('parseShardPair', () => { + it.each([ + { value: '1/1', shardIndex: 1, shardCount: 1 }, + { value: '1/4', shardIndex: 1, shardCount: 4 }, + { value: '4/4', shardIndex: 4, shardCount: 4 }, + { value: '10/128', shardIndex: 10, shardCount: 128 }, + { value: '1/9007199254740991', shardIndex: 1, shardCount: Number.MAX_SAFE_INTEGER } + ])('parses $value', ({ value, shardIndex, shardCount }) => { + expect(parseShardPair(value)).toEqual({ shardIndex, shardCount }); + }); + + it('parses the index as 1-based, matching Rush and every mainstream runner', () => { + // If this ever became 0-based, shard "1/4" would silently run the second partition and + // no partition would run the first, so the suite would pass while skipping tests. + expect(parseShardPair('1/4').shardIndex).toBe(1); + }); + + it.each([ + { description: 'an empty string', value: '' }, + { description: 'a bare index', value: '2' }, + { description: 'a missing count', value: '2/' }, + { description: 'a missing index', value: '/4' }, + { description: 'a non-numeric count', value: '2/x' }, + { description: 'a negative index', value: '-2/4' }, + { description: 'a fractional index', value: '2.5/4' }, + { description: 'surrounding whitespace', value: ' 2/4 ' }, + { description: 'a trailing newline', value: '2/4\n' }, + { description: 'a trailing carriage return', value: '2/4\r' }, + { description: 'internal whitespace', value: '2 / 4' }, + { description: 'a backslash separator', value: '2\\4' } + ])('rejects $description', ({ value }) => { + expect(() => parseShardPair(value)).toThrow(/expected the format/); + }); + + it.each([ + { description: 'a leading junk segment', value: 'x/1/2' }, + { description: 'a trailing junk segment', value: '1/2/x' }, + { description: 'three numeric segments', value: '1/2/3' } + ])('rejects $description rather than silently dropping it', ({ value }) => { + // Jest's own parser splits on "/" and filters out the non-numeric segments before + // checking that two remain, so it accepts "x/1/2" as shard 1 of 2. Accepting a typo as + // a valid partition would run a fraction of the suite and report success. + expect(() => parseShardPair(value)).toThrow(/expected the format/); + }); + + it.each([ + { description: 'a zero index', value: '0/4' }, + { description: 'a zero count', value: '0/0' }, + { description: 'a zero count with a valid index', value: '1/0' } + ])('rejects $description because the syntax is 1-based', ({ value }) => { + expect(() => parseShardPair(value)).toThrow(/1-based/); + }); + + it('rejects an index beyond the count', () => { + expect(() => parseShardPair('5/4')).toThrow( + 'Invalid "--shard" value "5/4": the index 5 is greater than the count 4.' + ); + }); + + it('names the offending value so the message is actionable', () => { + expect(() => parseShardPair('two/four')).toThrow('"two/four"'); + }); + + it.each(['1/9007199254740992', '9007199254740992/9007199254740992', `1/${'9'.repeat(400)}`])( + 'rejects unsafe integers: %s', + (value: string) => { + expect(() => parseShardPair(value)).toThrow(/safe integers/); + } + ); +}); + +describe('formatShardPair', () => { + it.each(['1/1', '2/4', '10/128'])('round-trips %s', (value: string) => { + expect(formatShardPair(parseShardPair(value))).toBe(value); + }); +}); + +describe('getShardFolderName', () => { + it('derives a distinct folder per shard', () => { + const names: string[] = [1, 2, 3, 4].map(getShardFolderName); + expect(names).toEqual(['shard-1', 'shard-2', 'shard-3', 'shard-4']); + expect(new Set(names).size).toBe(names.length); + }); + + describe('getShardTempFolderPath', () => { + it('isolates both the shard index and partition count', () => { + const root: string = path.resolve('temp'); + expect(getShardTempFolderPath(root, parseShardPair('2/4'))).toBe(path.join(root, 'shard-2-of-4')); + expect(getShardTempFolderPath(root, parseShardPair('2/3'))).toBe(path.join(root, 'shard-2-of-3')); + expect(getShardTempFolderPath(root, parseShardPair('1/4'))).toBe(path.join(root, 'shard-1-of-4')); + }); + }); +}); diff --git a/apps/heft/src/sharding/test/ShardOutputOwnership.test.ts b/apps/heft/src/sharding/test/ShardOutputOwnership.test.ts new file mode 100644 index 0000000000..3764c91e3b --- /dev/null +++ b/apps/heft/src/sharding/test/ShardOutputOwnership.test.ts @@ -0,0 +1,625 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import fs from 'node:fs/promises'; +import fsSync from 'node:fs'; +import path from 'node:path'; +import { execFileSync, spawnSync, type SpawnSyncReturns } from 'node:child_process'; + +import { FileSystem } from '@rushstack/node-core-library'; + +import { + acquireShardOutputLocksAsync, + getShardOutputIdentity, + releaseShardOutputLocks, + resolveShardOutputPathAsync, + resolveShardPathAsync, + SHARD_OUTPUT_OWNERSHIP_FOLDER_NAME, + type IShardOutputLock +} from '../ShardOutputOwnership'; +import { writeShardFileAtomicallyAsync } from '../ShardFileSystem'; + +const compiledFolder: string = path.dirname(__dirname); +const shortCompiledFolderName: string | undefined = + process.platform === 'win32' + ? execFileSync( + 'powershell.exe', + [ + '-NoProfile', + '-NonInteractive', + '-Command', + '(New-Object -ComObject Scripting.FileSystemObject).GetFolder($env:HEFT_SHARD_ALIAS_DIRECTORY).ShortName' + ], + { + encoding: 'utf8', + env: { ...process.env, HEFT_SHARD_ALIAS_DIRECTORY: compiledFolder } + } + ).trim() + : undefined; +const hasShortPathAlias: boolean = + !!shortCompiledFolderName && + shortCompiledFolderName.toLowerCase() !== path.basename(compiledFolder).toLowerCase(); + +function setDirectoryCaseSensitivity(folderPath: string, sensitive: boolean): boolean { + const result: SpawnSyncReturns = + process.platform === 'win32' + ? spawnSync( + 'fsutil.exe', + ['file', 'setCaseSensitiveInfo', folderPath, sensitive ? 'enable' : 'disable'], + { + encoding: 'utf8' + } + ) + : spawnSync('chattr', [sensitive ? '-F' : '+F', folderPath], { encoding: 'utf8' }); + if (result.error && (!('code' in result.error) || result.error.code !== 'ENOENT')) throw result.error; + return result.status === 0; +} + +function supportsMixedDirectoryCaseSensitivity(): boolean { + if (process.platform !== 'win32' && process.platform !== 'linux') return false; + const root: string = fsSync.mkdtempSync(path.join(__dirname, 'shard-case-capability-')); + try { + if (!setDirectoryCaseSensitivity(root, true)) return false; + const insensitive: string = path.join(root, 'insensitive'); + fsSync.mkdirSync(insensitive); + if (!setDirectoryCaseSensitivity(insensitive, false)) return false; + fsSync.writeFileSync(path.join(root, 'probe'), ''); + fsSync.writeFileSync(path.join(insensitive, 'probe'), ''); + return !fsSync.existsSync(path.join(root, 'PROBE')) && fsSync.existsSync(path.join(insensitive, 'PROBE')); + } finally { + fsSync.rmSync(root, { recursive: true, force: true }); + } +} + +const hasMixedDirectoryCaseSensitivity: boolean = supportsMixedDirectoryCaseSensitivity(); + +describe('Heft output claims and attempt fences', () => { + let folder: string; + let output: string; + const locks: IShardOutputLock[] = []; + + beforeEach(async () => { + folder = await fs.mkdtemp(path.join(__dirname, 'shard-claims-')); + output = path.join(folder, 'result.json'); + }); + afterEach(async () => { + jest.restoreAllMocks(); + for (const lock of locks) if (!lock.isReleased) lock.release(); + locks.length = 0; + await fs.rm(folder, { recursive: true, force: true }); + }); + + async function acquireAsync(signal?: AbortSignal): Promise { + const [lock] = await acquireShardOutputLocksAsync( + folder, + [output], + { project: 'example', phase: 'test', task: 'jest', shardIndex: 2, shardCount: 4, generation: 'G' }, + signal + ); + locks.push(lock); + return lock; + } + + it('reports the current holder without stealing its output claim', async () => { + const lock = await acquireAsync(); + const metadata = JSON.parse(await fs.readFile(lock.filePath, 'utf8')); + expect(metadata).toMatchObject({ + version: 2, + kind: 'output', + destination: 'result.json', + project: 'example', + phase: 'test', + task: 'jest', + shardIndex: 2, + shardCount: 4, + generation: 'G', + attemptId: lock.attemptId, + pid: process.pid, + host: expect.any(String), + acquiredAt: expect.any(String), + processStartedAt: expect.any(String) + }); + await expect(acquireAsync()).rejects.toThrow(lock.attemptId); + await expect(acquireAsync()).rejects.toThrow(/state=contended retryCount=0.*claimAgeMs/); + expect(await fs.readFile(lock.filePath, 'utf8')).toBe(JSON.stringify(metadata)); + }); + + it('resolves missing inputs without materializing them and outputs without creating result files', async () => { + const target: string = path.join(folder, 'new-parent', 'report.json'); + expect(await resolveShardPathAsync(target)).toBe(target); + expect(await fs.readdir(folder)).toEqual([]); + expect(await resolveShardOutputPathAsync(folder, target, false)).toBe(target); + expect(await fs.readdir(path.dirname(target))).toEqual([]); + await expect(resolveShardOutputPathAsync(folder, folder, true)).rejects.toThrow(/below the build folder/); + await expect( + resolveShardOutputPathAsync(folder, path.resolve(folder, '..', 'escape'), false) + ).rejects.toThrow(/below the build folder/); + await expect( + resolveShardOutputPathAsync( + folder, + path.join(folder, SHARD_OUTPUT_OWNERSHIP_FOLDER_NAME, 'report.json'), + false + ) + ).rejects.toThrow(/reserved/); + }); + + it('shares canonical output claims through a symlinked workspace root', async () => { + const alias: string = path.join(folder, 'workspace-alias'); + await fs.symlink(folder, alias, 'junction'); + const canonical: string = await resolveShardOutputPathAsync( + alias, + path.join(alias, 'result.json'), + false + ); + expect(canonical).toBe(output); + locks.push(...(await acquireShardOutputLocksAsync(alias, [canonical]))); + await expect(acquireAsync()).rejects.toThrow(/already owned/); + }); + + (hasShortPathAlias ? it : it.skip).each([false, true])( + 'canonicalizes Windows short-path ancestors (output already exists: %s)', + async (exists) => { + const buildFolder: string = path.dirname(compiledFolder); + const target: string = path.join(folder, 'new-parent', 'report.json'); + const alias: string = path.join( + buildFolder, + shortCompiledFolderName!, + path.relative(compiledFolder, target) + ); + if (exists) { + await fs.mkdir(path.dirname(target), { recursive: true }); + await fs.writeFile(target, 'preserve'); + } + const canonical: string = await resolveShardOutputPathAsync(buildFolder, target, false); + expect(await resolveShardOutputPathAsync(buildFolder, alias, false)).toBe(canonical); + // Canonicalization must not replace an existing report or create an absent one. + if (exists) expect(await fs.readFile(target, 'utf8')).toBe('preserve'); + else expect(await FileSystem.existsAsync(target)).toBe(false); + } + ); + + it('normalizes Unicode composition without approximating filesystem case folding', () => { + const composed: string = path.join(folder, 'reports', '\u00e9.json'); + const decomposed: string = path.join(folder, 'reports', 'e\u0301.json'); + expect(getShardOutputIdentity(folder, composed)).toBe(getShardOutputIdentity(folder, decomposed)); + expect(getShardOutputIdentity(folder, path.join(folder, 'Reports', '\u00e9.json'))).not.toBe( + getShardOutputIdentity(folder, composed) + ); + }); + + it.each([ + ['report.json', 'REPORT.JSON'], + ['\u00df.json', '\u1e9e.json'] + ])('uses filesystem name equivalence for missing leaves: %s / %s', async (first, second) => { + const parent: string = path.join(folder, 'reports'); + const firstPath: string = await resolveShardOutputPathAsync(folder, path.join(parent, first), false); + const secondPath: string = await resolveShardOutputPathAsync(folder, path.join(parent, second), false); + const probe: string = path.join(parent, first); + await fs.writeFile(probe, ''); + const aliases: boolean = await FileSystem.existsAsync(path.join(parent, second)); + await fs.unlink(probe); + locks.push(...(await acquireShardOutputLocksAsync(folder, [firstPath]))); + if (aliases) { + await expect(acquireShardOutputLocksAsync(folder, [secondPath])).rejects.toThrow(/already owned/); + } else { + const secondLocks: IShardOutputLock[] = await acquireShardOutputLocksAsync(folder, [secondPath]); + locks.push(...secondLocks); + expect(secondLocks[0].filePath).not.toBe(locks[0].filePath); + } + }); + + it.each([true, false])( + 'uses destination-parent semantics, not lock-directory semantics (modeled case-sensitive output: %s)', + async (sensitive) => { + const parent: string = path.join(folder, 'reports'); + const first: string = await resolveShardOutputPathAsync( + folder, + path.join(parent, 'result.json'), + false + ); + const second: string = await resolveShardOutputPathAsync( + folder, + path.join(parent, 'RESULT.JSON'), + false + ); + const lockFolder: string = path.join(folder, SHARD_OUTPUT_OWNERSHIP_FOLDER_NAME); + const realpath: typeof fs.realpath = fs.realpath; + jest.spyOn(fs, 'realpath').mockImplementation(async (filename, options) => { + if (filename === path.join(folder, SHARD_OUTPUT_OWNERSHIP_FOLDER_NAME.toUpperCase())) { + if (sensitive) return lockFolder; + throw Object.assign(new Error('case-sensitive lock directory'), { + code: 'ENOENT', + errno: -2, + syscall: 'realpath' + }); + } + return realpath(filename, options); + }); + const lstat: typeof fs.lstat = fs.lstat; + jest.spyOn(fs, 'lstat').mockImplementation(async (filename, options) => { + if (typeof filename === 'string' && filename.toLowerCase().includes('.heft-shard-name-probe-')) { + // Model opposite lookup rules in reports and lock storage; this is not native filesystem evidence. + const aliases: boolean = filename.startsWith(`${parent}${path.sep}`) ? !sensitive : sensitive; + const names: string[] = await fs.readdir(path.dirname(filename)); + const actual: string | undefined = names.find( + (name) => name.toLowerCase() === path.basename(filename).toLowerCase() + ); + if (actual && (aliases || actual === path.basename(filename))) { + return lstat(path.join(path.dirname(filename), actual), options); + } + throw Object.assign(new Error('case-distinct name'), { + code: 'ENOENT', + errno: -2, + syscall: 'lstat' + }); + } + return lstat(filename, options); + }); + const firstLocks: IShardOutputLock[] = await acquireShardOutputLocksAsync(folder, [first]); + locks.push(...firstLocks); + const acquireSecondAsync = async (): Promise => { + const acquired: IShardOutputLock[] = await acquireShardOutputLocksAsync(folder, [second]); + locks.push(...acquired); + return acquired; + }; + if (sensitive) { + expect(await acquireSecondAsync()).toHaveLength(1); + } else { + await expect(acquireSecondAsync()).rejects.toThrow(/already owned/); + } + releaseShardOutputLocks(locks); + if (sensitive) { + const acquired: IShardOutputLock[] = await acquireShardOutputLocksAsync(folder, [first, second]); + locks.push(...acquired); + expect(acquired).toHaveLength(2); + } else { + await expect(acquireShardOutputLocksAsync(folder, [first, second])).rejects.toThrow( + /overlap within one operation/ + ); + } + expect(await fs.readdir(parent)).toEqual([]); + } + ); + + (hasMixedDirectoryCaseSensitivity ? it : it.skip).each([ + { sensitive: true, basename: 'result.json' }, + { sensitive: false, basename: 'result.json' }, + { sensitive: true, basename: 'r'.repeat(255) }, + { sensitive: false, basename: 'r'.repeat(255) } + ])( + 'uses native mixed directory semantics for absent outputs (case-sensitive output: $sensitive, $basename)', + async ({ sensitive, basename }) => { + expect(setDirectoryCaseSensitivity(folder, !sensitive)).toBe(true); + const parent: string = path.join(folder, 'reports'); + await fs.mkdir(parent); + expect(setDirectoryCaseSensitivity(parent, sensitive)).toBe(true); + await fs.writeFile(path.join(parent, 'probe'), ''); + expect(await FileSystem.existsAsync(path.join(parent, 'PROBE'))).toBe(!sensitive); + await fs.unlink(path.join(parent, 'probe')); + const first: string = await resolveShardOutputPathAsync(folder, path.join(parent, basename), false); + const second: string = await resolveShardOutputPathAsync( + folder, + path.join(parent, basename.toUpperCase()), + false + ); + locks.push(...(await acquireShardOutputLocksAsync(folder, [first]))); + const acquireSecondAsync = async (): Promise => { + const acquired: IShardOutputLock[] = await acquireShardOutputLocksAsync(folder, [second]); + locks.push(...acquired); + return acquired; + }; + if (sensitive) { + expect(await acquireSecondAsync()).toHaveLength(1); + } else { + await expect(acquireSecondAsync()).rejects.toThrow(/already owned/); + } + releaseShardOutputLocks(locks); + if (!sensitive) { + await expect(acquireShardOutputLocksAsync(folder, [first, second])).rejects.toThrow( + /overlap within one operation/ + ); + } + expect(await fs.readdir(parent)).toEqual([]); + } + ); + + (hasMixedDirectoryCaseSensitivity ? it : it.skip).each([true, false])( + 'rejects a probe that does not inherit actual-parent case sensitivity (sensitive parent: %s)', + async (sensitive) => { + const parent: string = path.join(folder, 'reports'); + await fs.mkdir(parent); + expect(setDirectoryCaseSensitivity(parent, sensitive)).toBe(true); + const first: string = await resolveShardOutputPathAsync(folder, path.join(parent, 'first.json'), false); + const second: string = await resolveShardOutputPathAsync( + folder, + path.join(parent, 'second.json'), + false + ); + locks.push(...(await acquireShardOutputLocksAsync(folder, [first]))); + const mkdtemp: typeof fs.mkdtemp = fs.mkdtemp; + jest.spyOn(fs, 'mkdtemp').mockImplementation(async (prefix, options) => { + const probePath: string | Buffer = await mkdtemp(prefix, options); + if (typeof probePath !== 'string') throw new Error('Expected a string probe path'); + expect(setDirectoryCaseSensitivity(probePath, !sensitive)).toBe(true); + return probePath; + }); + await expect(acquireShardOutputLocksAsync(folder, [second])).rejects.toThrow( + /probe directory does not preserve the case sensitivity/ + ); + expect(await fs.readdir(parent)).toEqual([]); + expect(await fs.readdir(path.join(folder, SHARD_OUTPUT_OWNERSHIP_FOLDER_NAME))).toEqual([ + path.basename(locks[0].filePath) + ]); + } + ); + + it('uses the physical spelling of a Unicode-normalized destination parent', async () => { + const parent: string = path.join(folder, 'e\u0301'); + const first: string = await resolveShardOutputPathAsync(folder, path.join(parent, 'first.json'), false); + const second: string = await resolveShardOutputPathAsync(folder, path.join(parent, 'second.json'), false); + locks.push(...(await acquireShardOutputLocksAsync(folder, [first, second]))); + expect(locks).toHaveLength(2); + expect(await fs.readdir(parent)).toEqual([]); + }); + + it.each( + [ + { description: '200 ASCII characters', prefix: 'a'.repeat(199) }, + { description: '255 ASCII characters', prefix: 'a'.repeat(254) }, + { + description: 'near-limit Unicode characters', + prefix: '\u00e9'.repeat(process.platform === 'win32' ? 254 : 127) + } + ].flatMap(({ description, prefix }) => [ + { description, prefix, ancestor: false }, + { description, prefix, ancestor: true } + ]) + )('claims valid long components ($description, ancestor: $ancestor)', async ({ prefix, ancestor }) => { + const first: string = await resolveShardOutputPathAsync( + folder, + ancestor ? path.join(folder, `${prefix}1`, 'result.json') : path.join(folder, 'reports', `${prefix}1`), + false + ); + const second: string = await resolveShardOutputPathAsync( + folder, + ancestor ? path.join(folder, `${prefix}2`, 'result.json') : path.join(folder, 'reports', `${prefix}2`), + false + ); + // Establish that these complete names are supported, then claim them while the result files are absent. + await fs.writeFile(first, 'first'); + await fs.writeFile(second, 'second'); + expect(await fs.readFile(first, 'utf8')).toBe('first'); + expect(await fs.readFile(second, 'utf8')).toBe('second'); + await fs.unlink(first); + await fs.unlink(second); + locks.push(...(await acquireShardOutputLocksAsync(folder, [first]))); + locks.push(...(await acquireShardOutputLocksAsync(folder, [second]))); + expect(locks).toHaveLength(2); + releaseShardOutputLocks(locks); + const acquired: IShardOutputLock[] = await acquireShardOutputLocksAsync(folder, [first, second]); + locks.push(...acquired); + expect(acquired).toHaveLength(2); + expect(await fs.readdir(path.dirname(first))).toEqual([]); + expect(await fs.readdir(path.dirname(second))).toEqual([]); + expect((await fs.readdir(folder)).filter((name) => name.startsWith('.heft-shard-name-probe-'))).toEqual( + [] + ); + }); + + it('fails closed if a destination parent disappears before a name comparison', async () => { + const parent: string = path.join(folder, 'reports'); + const first: string = await resolveShardOutputPathAsync(folder, path.join(parent, 'first.json'), false); + const second: string = await resolveShardOutputPathAsync(folder, path.join(parent, 'second.json'), false); + locks.push(...(await acquireShardOutputLocksAsync(folder, [first]))); + await fs.rmdir(parent); + await expect( + acquireShardOutputLocksAsync(folder, [second]).then((acquired) => { + locks.push(...acquired); + return acquired; + }) + ).rejects.toThrow(/ENOENT/); + expect(await fs.readdir(path.join(folder, SHARD_OUTPUT_OWNERSHIP_FOLDER_NAME))).toEqual([ + path.basename(locks[0].filePath) + ]); + }); + + it('surfaces probe lookup errors, removes its probe and releases the registration gate', async () => { + const parent: string = path.join(folder, 'reports'); + const first: string = await resolveShardOutputPathAsync(folder, path.join(parent, 'first.json'), false); + const second: string = await resolveShardOutputPathAsync(folder, path.join(parent, 'second.json'), false); + locks.push(...(await acquireShardOutputLocksAsync(folder, [first]))); + const lstat: typeof fs.lstat = fs.lstat; + jest.spyOn(fs, 'lstat').mockImplementation(async (filename, options) => { + if (typeof filename === 'string' && filename.includes('.heft-shard-name-probe-')) { + throw Object.assign(new Error('probe lookup denied'), { code: 'EACCES' }); + } + return lstat(filename, options); + }); + await expect(acquireShardOutputLocksAsync(folder, [second])).rejects.toThrow('probe lookup denied'); + expect(await fs.readdir(parent)).toEqual([]); + expect(await fs.readdir(path.join(folder, SHARD_OUTPUT_OWNERSHIP_FOLDER_NAME))).toEqual([ + path.basename(locks[0].filePath) + ]); + }); + + it.each([false, true])('excludes ancestor and descendant claims (reverse: %s)', async (reverse) => { + const parent: string = path.join(folder, 'reports'); + const child: string = path.join(parent, 'nested', 'report.json'); + locks.push(...(await acquireShardOutputLocksAsync(folder, [reverse ? child : parent]))); + await expect(acquireShardOutputLocksAsync(folder, [reverse ? parent : child])).rejects.toThrow( + /already owned/ + ); + }); + + it('rejects overlapping outputs within one registration', async () => { + const parent: string = path.join(folder, 'reports'); + await expect( + acquireShardOutputLocksAsync(folder, [parent, path.join(parent, 'report.json')]) + ).rejects.toThrow(/overlap within one operation/); + expect(await fs.readdir(path.join(folder, SHARD_OUTPUT_OWNERSHIP_FOLDER_NAME))).toEqual([]); + }); + + it('registers disjoint claim sets concurrently and deduplicates exact destinations', async () => { + const [first, second] = await Promise.all([ + acquireShardOutputLocksAsync(folder, [output, output]), + acquireShardOutputLocksAsync(folder, [path.join(folder, 'other.json'), path.join(folder, 'coverage')]) + ]); + locks.push(...first, ...second); + expect(first).toHaveLength(1); + expect(second).toHaveLength(2); + expect(first[0].attemptId).not.toBe(second[0].attemptId); + expect(second[0].attemptId).toBe(second[1].attemptId); + expect(await fs.readdir(path.join(folder, SHARD_OUTPUT_OWNERSHIP_FOLDER_NAME))).toHaveLength(3); + releaseShardOutputLocks(locks); + expect(await fs.readdir(path.join(folder, SHARD_OUTPUT_OWNERSHIP_FOLDER_NAME))).toEqual([]); + }); + + it('releases all partial claims after a registration write error', async () => { + const writeFile: typeof fsSync.writeFileSync = fsSync.writeFileSync; + jest.spyOn(fsSync, 'writeFileSync').mockImplementation((filename, data, options) => { + if (typeof data === 'string' && JSON.parse(data).destination === 'two.json') { + throw new Error('claim registration denied'); + } + writeFile(filename, data, options); + }); + const paths: string[] = [output, path.join(folder, 'two.json')]; + await expect(acquireShardOutputLocksAsync(folder, paths)).rejects.toThrow('claim registration denied'); + jest.restoreAllMocks(); + expect(await fs.readdir(path.join(folder, SHARD_OUTPUT_OWNERSHIP_FOLDER_NAME))).toEqual([]); + locks.push(...(await acquireShardOutputLocksAsync(folder, paths))); + }); + + it('cleans a newly created lock when its filesystem identity cannot be read', async () => { + jest.spyOn(fsSync, 'fstatSync').mockImplementationOnce(() => { + throw new Error('identity unavailable'); + }); + await expect(acquireAsync()).rejects.toThrow('identity unavailable'); + expect(await fs.readdir(path.join(folder, SHARD_OUTPUT_OWNERSHIP_FOLDER_NAME))).toEqual([]); + await acquireAsync(); + }); + + it('releases registered outputs but preserves a replaced registration gate', async () => { + const gate: string = path.join(folder, SHARD_OUTPUT_OWNERSHIP_FOLDER_NAME, 'acquire.lock'); + const writeFile: typeof fsSync.writeFileSync = fsSync.writeFileSync; + jest.spyOn(fsSync, 'writeFileSync').mockImplementation((filename, data, options) => { + writeFile(filename, data, options); + if (typeof data === 'string' && JSON.parse(data).kind === 'output') { + fsSync.unlinkSync(gate); + writeFile(gate, 'replacement'); + } + }); + await expect(acquireAsync()).rejects.toThrow(/finalize managed shard output ownership registration/); + jest.restoreAllMocks(); + expect(await fs.readdir(path.dirname(gate))).toEqual(['acquire.lock']); + expect(await fs.readFile(gate, 'utf8')).toBe('replacement'); + }); + + it('does not unlink a claim that gained another hard link while owned', async () => { + const lock: IShardOutputLock = await acquireAsync(); + const link: string = path.join(folder, 'preserve-claim'); + await fs.link(lock.filePath, link); + const contents: string = await fs.readFile(link, 'utf8'); + expect(() => lock.release()).toThrow(/changed while in use/); + expect(lock.isReleased).toBe(true); + expect(await fs.readFile(lock.filePath, 'utf8')).toBe(contents); + await expect(acquireAsync()).rejects.toThrow(/already owned/); + }); + + it('rejects a junction at the ownership folder without changing its target', async () => { + const target: string = path.join(folder, 'other-locks'); + await fs.mkdir(target); + await fs.symlink(target, path.join(folder, SHARD_OUTPUT_OWNERSHIP_FOLDER_NAME), 'junction'); + await expect(acquireAsync()).rejects.toThrow(/symbolic link/); + expect(await fs.readdir(target)).toEqual([]); + }); + + it('rejects malformed or legacy claims without reclaiming them', async () => { + const lock: IShardOutputLock = await acquireAsync(); + lock.release(); + await fs.writeFile(lock.filePath, '{"version":1}'); + await expect(acquireAsync()).rejects.toThrow(/malformed or from an older runtime/); + expect(await fs.readFile(lock.filePath, 'utf8')).toBe('{"version":1}'); + }); + + it('revokes A before releasing its claim and rejects a late A promotion after B starts', async () => { + const controller = new AbortController(); + const first = await acquireAsync(controller.signal); + controller.abort(); + expect(first.state).toBe('revoked'); + await expect(acquireAsync()).rejects.toThrow(/already owned/); + first.release(); + const second = await acquireAsync(); + expect(second.attemptId).not.toBe(first.attemptId); + await writeShardFileAtomicallyAsync(output, 'B', undefined, second); + + // A queued continuation still carries A's capability, even if it loses the abort signal. + await expect(writeShardFileAtomicallyAsync(output, 'late A', undefined, first)).rejects.toThrow( + /released/ + ); + expect(await fs.readFile(output, 'utf8')).toBe('B'); + expect(await fs.readdir(folder)).toEqual( + expect.arrayContaining(['result.json', SHARD_OUTPUT_OWNERSHIP_FOLDER_NAME]) + ); + expect((await fs.readdir(folder)).filter((name) => name.endsWith('.tmp'))).toEqual([]); + }); + + it('does not release a claim between the final fence check and manifest promotion', async () => { + const lock = await acquireAsync(); + const rename = fsSync.renameSync; + jest.spyOn(fsSync, 'renameSync').mockImplementation((source, destination) => { + expect(lock.state).toBe('publishing'); + expect(() => lock.release()).toThrow(/Cannot release a publishing/); + rename(source, destination); + }); + await writeShardFileAtomicallyAsync(output, 'complete', undefined, lock); + expect(lock.state).toBe('held'); + expect(await fs.readFile(output, 'utf8')).toBe('complete'); + }); + + it('does not close a reused descriptor when a released claim is released again', async () => { + const first = await acquireAsync(); + first.release(); + const descriptor = fsSync.openSync(output, 'w'); + try { + expect(() => first.release()).toThrow(/already released/); + fsSync.writeSync(descriptor, 'still owned by the caller'); + } finally { + fsSync.closeSync(descriptor); + } + expect(await fs.readFile(output, 'utf8')).toBe('still owned by the caller'); + }); + + it('invalidates a just-promoted result when cancellation arrives before returning to the caller', async () => { + const controller = new AbortController(); + const lock = await acquireAsync(controller.signal); + const rename = fsSync.renameSync; + jest.spyOn(fsSync, 'renameSync').mockImplementation((source, destination) => { + rename(source, destination); + controller.abort(); + }); + await expect( + writeShardFileAtomicallyAsync(output, 'cancelled', controller.signal, lock) + ).rejects.toThrow(); + expect(lock.state).toBe('revoked'); + await expect(fs.stat(output)).rejects.toThrow(/ENOENT/); + await expect(acquireAsync()).rejects.toThrow(/already owned/); + }); + + it('reports bounded registration exhaustion and preserves an abandoned gate for verified cleanup', async () => { + const lock = await acquireAsync(); + const metadata = JSON.parse(await fs.readFile(lock.filePath, 'utf8')); + lock.release(); + const gate = path.join(folder, SHARD_OUTPUT_OWNERSHIP_FOLDER_NAME, 'acquire.lock'); + metadata.kind = 'registration'; + metadata.acquiredAt = '2000-01-01T00:00:00.000Z'; + const contents = JSON.stringify(metadata); + await fs.writeFile(gate, contents); + await expect(acquireAsync()).rejects.toThrow( + /state=retry-exhausted retryCount=\d+ retryDeadline=.*Never steal a claim based only on PID or age/ + ); + expect(await fs.readFile(gate, 'utf8')).toBe(contents); + // This test owns the synthetic gate; production recovery requires verified process termination. + await fs.unlink(gate); + expect((await acquireAsync()).state).toBe('held'); + }); +}); diff --git a/apps/heft/src/sharding/test/ShardResultManifest.test.ts b/apps/heft/src/sharding/test/ShardResultManifest.test.ts new file mode 100644 index 0000000000..003df6a6f9 --- /dev/null +++ b/apps/heft/src/sharding/test/ShardResultManifest.test.ts @@ -0,0 +1,218 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import fs from 'node:fs/promises'; +import fsSync from 'node:fs'; +import path from 'node:path'; + +import { FileSystem } from '@rushstack/node-core-library'; + +import type { IShardResultManifest } from '../IShardResultManifest'; +import { + SHARD_MANIFEST_FILENAME, + getShardManifestPath, + invalidateShardResultManifestAsync, + readShardResultManifestAsync, + tryReadShardResultManifestAsync, + validateShardResultManifest, + writeShardResultManifestAsync +} from '../ShardResultManifest'; + +function createManifest(overrides?: Partial): IShardResultManifest { + return { + version: 1, + shardIndex: 1, + shardCount: 4, + generation: 'abc123', + producerAttempt: 'attempt-1', + status: 'completed', + workload: 'jest', + startedAt: '2026-09-14T00:00:00.000Z', + durationMs: 1234, + diagnosticCount: 0, + itemCount: 7, + ...overrides + }; +} + +describe('ShardResultManifest', () => { + let shardFolderPath: string; + + beforeEach(async () => { + shardFolderPath = await fs.mkdtemp(path.join(__dirname, 'heft-shard-')); + }); + + afterEach(async () => { + jest.restoreAllMocks(); + await FileSystem.deleteFolderAsync(shardFolderPath); + }); + + it('places the manifest at the documented filename within the shard folder', () => { + expect(getShardManifestPath(shardFolderPath)).toEqual( + path.join(shardFolderPath, SHARD_MANIFEST_FILENAME) + ); + }); + + it('round-trips a manifest', async () => { + const manifest: IShardResultManifest = createManifest(); + await writeShardResultManifestAsync(shardFolderPath, manifest); + + await expect(readShardResultManifestAsync(shardFolderPath)).resolves.toEqual(manifest); + expect(await fs.readdir(shardFolderPath)).toEqual([SHARD_MANIFEST_FILENAME]); + }); + + it('validates an in-memory manifest without interpreting workload diagnostics', () => { + const manifest: IShardResultManifest = createManifest({ diagnosticCount: 3, itemCount: 0 }); + expect(validateShardResultManifest(manifest)).toBe(manifest); + expect(validateShardResultManifest(createManifest({ status: 'harness-failure' })).status).toBe( + 'harness-failure' + ); + }); + + it('invalidates only its own manifest, preserving other output and siblings', async () => { + await writeShardResultManifestAsync(shardFolderPath, createManifest()); + await fs.writeFile(path.join(shardFolderPath, 'report.json'), '{}'); + await invalidateShardResultManifestAsync(shardFolderPath); + await invalidateShardResultManifestAsync(shardFolderPath); + expect(await fs.readdir(shardFolderPath)).toEqual(['report.json']); + }); + + it('rejects invalid output before publishing a commit marker', async () => { + await expect( + writeShardResultManifestAsync(shardFolderPath, createManifest({ itemCount: -1 })) + ).rejects.toThrow(/JSON validation failed/); + expect(await fs.readdir(shardFolderPath)).toEqual([]); + }); + + it('propagates an atomic rename failure and removes only its own staging file', async () => { + await fs.mkdir(getShardManifestPath(shardFolderPath)); + await fs.writeFile(path.join(shardFolderPath, 'report.json'), '{}'); + await expect(writeShardResultManifestAsync(shardFolderPath, createManifest())).rejects.toThrow(); + expect((await fs.readdir(shardFolderPath)).sort()).toEqual(['report.json', SHARD_MANIFEST_FILENAME]); + }); + + it('does not publish a cancelled result', async () => { + const controller: AbortController = new AbortController(); + controller.abort(); + await expect( + writeShardResultManifestAsync(shardFolderPath, createManifest(), controller.signal) + ).rejects.toThrow(); + await expect(tryReadShardResultManifestAsync(shardFolderPath)).resolves.toBeUndefined(); + }); + + it('checks cancellation after awaiting the atomic writer', async () => { + const controller: AbortController = new AbortController(); + const originalRename: typeof fs.rename = fs.rename; + jest.spyOn(fs, 'rename').mockImplementation(async (...args) => { + await originalRename(...args); + queueMicrotask(() => + queueMicrotask(() => controller.abort(new Error('cancelled after atomic writer'))) + ); + }); + await expect( + writeShardResultManifestAsync(shardFolderPath, createManifest(), controller.signal) + ).rejects.toThrow('cancelled after atomic writer'); + await expect(tryReadShardResultManifestAsync(shardFolderPath)).resolves.toBeUndefined(); + }); + + it.each([1, 2])('does not remove a replacement during cancellation at await boundary %s', async (depth) => { + const controller: AbortController = new AbortController(); + const replacementPath: string = path.join(shardFolderPath, 'replacement.json'); + const replacement: IShardResultManifest = createManifest({ generation: 'replacement' }); + await fs.writeFile(replacementPath, JSON.stringify(replacement)); + const originalRename: typeof fs.rename = fs.rename; + jest.spyOn(fs, 'rename').mockImplementation(async (...args) => { + await originalRename(...args); + let remaining: number = depth; + const replaceAndCancel: () => void = () => { + if (--remaining > 0) { + queueMicrotask(replaceAndCancel); + } else { + fsSync.renameSync(replacementPath, getShardManifestPath(shardFolderPath)); + controller.abort(new Error('cancelled original publication')); + } + }; + queueMicrotask(replaceAndCancel); + }); + await expect( + writeShardResultManifestAsync(shardFolderPath, createManifest(), controller.signal) + ).rejects.toThrow('cancelled original publication'); + expect(await readShardResultManifestAsync(shardFolderPath)).toEqual(replacement); + expect(await fs.readdir(shardFolderPath)).toEqual([SHARD_MANIFEST_FILENAME]); + }); + + it('preserves an adapter payload without inspecting it', async () => { + const payload: unknown = { testResults: [{ name: 'a' }], nested: { deep: [1, 2, 3] } }; + await writeShardResultManifestAsync(shardFolderPath, createManifest({ payload })); + + const result: IShardResultManifest = await readShardResultManifestAsync(shardFolderPath); + expect(result.payload).toEqual(payload); + }); + + it('writes a manifest whose payload was explicitly set to undefined', async () => { + await writeShardResultManifestAsync(shardFolderPath, createManifest({ payload: undefined })); + + const result: IShardResultManifest = await readShardResultManifestAsync(shardFolderPath); + expect(result.payload).toBeUndefined(); + }); + + it('creates the shard folder if it does not yet exist', async () => { + const nestedFolderPath: string = path.join(shardFolderPath, 'does', 'not', 'exist'); + await writeShardResultManifestAsync(nestedFolderPath, createManifest()); + + await expect(readShardResultManifestAsync(nestedFolderPath)).resolves.toBeDefined(); + }); + + it('reports a missing manifest as undefined rather than throwing', async () => { + await expect(tryReadShardResultManifestAsync(shardFolderPath)).resolves.toBeUndefined(); + }); + + it('throws when a required manifest is missing', async () => { + await expect(readShardResultManifestAsync(shardFolderPath)).rejects.toThrow(/ENOENT|no such file/i); + }); + + it('throws rather than reporting undefined when a manifest is present but unparseable', async () => { + await FileSystem.writeFileAsync(getShardManifestPath(shardFolderPath), '{ not json'); + + await expect(tryReadShardResultManifestAsync(shardFolderPath)).rejects.toThrow(/Error reading/); + }); + + it('throws rather than reporting undefined when a manifest is present but invalid', async () => { + await FileSystem.writeFileAsync(getShardManifestPath(shardFolderPath), JSON.stringify({ version: 1 })); + + await expect(tryReadShardResultManifestAsync(shardFolderPath)).rejects.toThrow(/JSON validation failed/); + }); + + it.each([ + { description: 'an unsupported version', overrides: { version: 2 } }, + { description: 'a zero shard index', overrides: { shardIndex: 0 } }, + { description: 'a zero shard count', overrides: { shardCount: 0 } }, + { description: 'a fractional shard index', overrides: { shardIndex: 1.5 } }, + { description: 'an unsafe shard index', overrides: { shardIndex: Number.MAX_SAFE_INTEGER + 1 } }, + { description: 'an empty generation', overrides: { generation: '' } }, + { description: 'a missing generation', overrides: { generation: undefined } }, + { description: 'an empty attempt', overrides: { producerAttempt: '' } }, + { description: 'a missing attempt', overrides: { producerAttempt: undefined } }, + { description: 'an empty workload', overrides: { workload: '' } }, + { description: 'an unrecognized status', overrides: { status: 'ok' } }, + { description: 'a non-timestamp startedAt', overrides: { startedAt: 'not a date' } }, + { description: 'a negative duration', overrides: { durationMs: -1 } }, + { description: 'a non-finite duration', overrides: { durationMs: Number.POSITIVE_INFINITY } }, + { description: 'a negative diagnostic count', overrides: { diagnosticCount: -1 } }, + { description: 'a fractional item count', overrides: { itemCount: 1.5 } }, + { description: 'an unsafe item count', overrides: { itemCount: Number.MAX_SAFE_INTEGER + 1 } }, + { + description: 'an unsafe diagnostic count', + overrides: { diagnosticCount: Number.MAX_SAFE_INTEGER + 1 } + }, + { description: 'an unsafe shard count', overrides: { shardCount: Number.MAX_SAFE_INTEGER + 1 } }, + { description: 'an unrecognized property', overrides: { unexpected: true } } + ])('rejects a manifest with $description', async ({ overrides }) => { + await FileSystem.writeFileAsync( + getShardManifestPath(shardFolderPath), + JSON.stringify({ ...createManifest(), ...overrides }) + ); + + await expect(readShardResultManifestAsync(shardFolderPath)).rejects.toThrow(/JSON validation failed/); + }); +}); diff --git a/common/changes/@rushstack/heft/heft-shard-contracts_2026-11-13.json b/common/changes/@rushstack/heft/heft-shard-contracts_2026-11-13.json new file mode 100644 index 0000000000..f105299e7c --- /dev/null +++ b/common/changes/@rushstack/heft/heft-shard-contracts_2026-11-13.json @@ -0,0 +1,10 @@ +{ + "changes": [ + { + "packageName": "@rushstack/heft", + "comment": "Add workload-neutral shard identities, result manifests, atomic file publication and exclusive output claims.", + "type": "minor" + } + ], + "packageName": "@rushstack/heft" +} diff --git a/common/reviews/api/heft.api.md b/common/reviews/api/heft.api.md index 89c0f114e6..3acdbe0fef 100644 --- a/common/reviews/api/heft.api.md +++ b/common/reviews/api/heft.api.md @@ -390,6 +390,22 @@ export interface IScopedLogger { readonly terminal: ITerminal; } +// @public +export interface IShardResultManifest { + diagnosticCount: number; + durationMs: number; + generation: string; + itemCount: number; + payload?: unknown; + producerAttempt: string; + shardCount: number; + shardIndex: number; + startedAt: string; + status: ShardStatus; + version: 1; + workload: string; +} + // @public export interface IWatchedFileState { changed: boolean; @@ -436,6 +452,9 @@ export type ReaddirDirentCallback = (error: NodeJS.ErrnoException | null, files: // @public export type ReaddirStringCallback = (error: NodeJS.ErrnoException | null, files: string[]) => void; +// @public +export type ShardStatus = 'completed' | 'harness-failure'; + // @public export type StatCallback = (error: NodeJS.ErrnoException | null, stats: fs.Stats) => void; diff --git a/docs/heft/shard-contracts.md b/docs/heft/shard-contracts.md new file mode 100644 index 0000000000..37b6205121 --- /dev/null +++ b/docs/heft/shard-contracts.md @@ -0,0 +1,73 @@ +# Heft shard contracts + +These are workload-neutral building blocks, not an enabled sharding mode. They do not schedule shards, +register collators, interpret Jest results, merge reports, or integrate with Rush caching. + +## Identity and manifest + +`parseShardPair()` accepts a strict `index/count` pair of positive safe integers with `index <= count`. +Indexes are 1-based. Formatting and folder helpers consume validated identities; shard temp folders include +both index and count (`shard-2-of-4`) so a different partition does not reuse the same temp state. + +`IShardResultManifest` and `ShardStatus` are exported from `@rushstack/heft`. The helpers are available through +`@rushstack/heft/lib/sharding/ShardIdentity`, `ShardResultManifest`, `ShardFileSystem` and `ShardOutputOwnership`. +The version 1 `shard-manifest.json` envelope contains: + +| Fields | Contract | +| --- | --- | +| `shardIndex`, `shardCount` | Positive safe integers identifying the partition. | +| `generation` | Nonempty semantic run identity supplied by the caller. | +| `producerAttempt` | Nonempty live-attempt identity from the acquired output claim, not a cache key. | +| `status` | `completed` includes completed work with diagnostics; `harness-failure` is not completion evidence. | +| `workload`, `startedAt`, `durationMs` | Nonempty workload name, ISO timestamp and nonnegative finite duration. | +| `diagnosticCount`, `itemCount` | Nonnegative safe integers; an empty partition is representable. | +| `payload` | Optional JSON-serializable workload data, opaque to the generic schema. | + +Unknown versions, extra envelope fields and invalid field domains fail schema validation on read and write. +`tryReadShardResultManifestAsync()` returns `undefined` only for absence; malformed or corrupt data still throws. +Serialization is validated before publication. Schema validity alone does not establish a complete or current +run: consumers must check index/count agreement, missing or duplicate siblings, current expected generation, +workload/status agreement, aggregate count overflow and payload integrity before merging results. Generation +derivation, sibling coherence checks and workload payload validation are outside this foundation. + +## Paths and exclusive ownership + +Resolve build-root-relative configuration paths against the build root before calling the path helpers. +Outputs must be strict descendants of that root. Validation rejects traversal outside the root and symbolic +links/junctions below it, while allowing a symlinked workspace root. `resolveShardPathAsync()` resolves existing +ancestors without creating missing inputs; `resolveShardOutputPathAsync()` validates and materializes output +directories, but never creates a result file. It resolves physical aliases, including Windows short paths. + +Pass canonical destinations to `acquireShardOutputLocksAsync()`. Claims exclude overlapping ancestors and +descendants, including overlap within one acquisition. Unicode composition is normalized; missing-name case +equivalence is probed beneath each differing component's actual parent, never the lock directory. A temporary +directory holds complete, unmodified names so probes do not exceed component-length limits. Its case sensitivity +is checked against the actual parent before comparing names; non-inherited case settings fail explicitly. +Probes require existing, writable destination parents; call `resolveShardOutputPathAsync()` first. If a parent +disappears or a probe cannot be created, inspected or removed, acquisition fails rather than guessing. +A short registration gate serializes multi-output acquisition, not workload execution. Disjoint claim sets +can remain held concurrently. +Output contention fails immediately; registration contention has a two-second retry budget, not an I/O timeout. +Partial acquisition failures release acquired claims and surface cleanup failures. + +Claims live in the reserved `/.heft-jest-shard-locks` protocol namespace, outside shard outputs. +Despite its historical name, this namespace is workload-neutral. Never cache or broadly clean it. Claims record +the destination, attempt, host, PID, timestamps and optional operation identity for actionable contention errors. +They are not leases: age, PID reuse or malformed metadata never authorize automatic recovery. Remove only the +specific abandoned claim after verifying that its owning process, workers and pending writes have all stopped. + +The caller must designate disjoint input/output locations, reject outputs overlapping source or payload inputs, +and retain ownership through all writes and cancellation cleanup. These helpers are cooperative process +coordination, not an OS sandbox against arbitrary writers or concurrent filesystem mutation. + +## Publication lifecycle + +Acquire ownership before invalidating a previous result. Invalidate only the known output file, publish payloads +first, then publish the manifest as the final commit marker. The atomic writer creates an exclusive staging +file beside its destination, writes and flushes it, closes it, and renames it. Pass the held claim as the +publication guard so ownership checking and promotion run synchronously without an intervening await. + +Cancellation revokes publication but retains the physical claim until work and cleanup settle. A cancelled +publication removes only the file identity it published, never a replacement. Failed publication cleans its +own staging file without deleting unrelated outputs. Release is synchronous; released or revoked claims cannot +publish again. The caller must perform its final cancellation check before releasing ownership.