diff --git a/packages/integration-sdk-runtime/src/__tests__/fileSystem.test.ts b/packages/integration-sdk-runtime/src/__tests__/fileSystem.test.ts new file mode 100644 index 000000000..708ce500b --- /dev/null +++ b/packages/integration-sdk-runtime/src/__tests__/fileSystem.test.ts @@ -0,0 +1,147 @@ +import { promises as fs } from 'fs'; + +import { vol } from 'memfs'; + +import { + OUT_OF_DISK_EXIT_CODE, + symlink, + writeFileToPath, + writeJsonToPath, +} from '../fileSystem'; + +// The shared `__mocks__/fs` re-exports memfs with `export *`, which makes every +// property non-configurable and therefore impossible to spy on. Rebuilding it +// here as a plain object keeps memfs behaviour while leaving `writeSync` +// replaceable, since that is the call the diagnostic goes through. +jest.mock('fs', () => { + const memfs = jest.requireActual('memfs'); + return { + ...memfs, + readdirSync: jest.requireActual('fs').readdirSync, + writeSync: jest.fn(), + }; +}); + +const mockedFs = jest.requireMock('fs') as { writeSync: jest.Mock }; + +function outOfDiskError() { + return Object.assign( + new Error("ENOSPC: no space left on device, write '/tmp/whatever'"), + { code: 'ENOSPC', errno: -28, syscall: 'write' }, + ); +} + +/** + * `process.exit` would take the jest worker down with it, so every test that + * can reach it replaces it with a no-op. Execution then falls through to the + * `throw` that follows, which is what the assertions below account for. + */ +function mockProcessExit() { + return jest.spyOn(process, 'exit').mockImplementation((() => { + /* keep the worker alive */ + }) as never); +} + +afterEach(() => { + vol.reset(); + // `clearMocks` only clears recorded calls, so an implementation installed by + // one test would otherwise leak into the next. + mockedFs.writeSync.mockReset(); +}); + +test('claims the exit code the ECS state machine treats as "retry on a bigger disk"', () => { + // Changing this breaks the contract with `handleTaskFailure` in + // jupiter-integration-service, which is deployed separately. + expect(OUT_OF_DISK_EXIT_CODE).toBe(77); +}); + +test('exits with the out-of-disk code when a file write runs out of space', async () => { + const exit = mockProcessExit(); + jest.spyOn(fs, 'writeFile').mockRejectedValue(outOfDiskError()); + + await expect( + writeFileToPath({ path: 'graph/entities/a.json', content: '{}' }), + ).rejects.toThrow('ENOSPC'); + + expect(exit).toHaveBeenCalledWith(OUT_OF_DISK_EXIT_CODE); +}); + +test('exits with the out-of-disk code when the storage directory cannot be created', async () => { + const exit = mockProcessExit(); + jest.spyOn(fs, 'mkdir').mockRejectedValue(outOfDiskError()); + + await expect( + writeFileToPath({ path: 'graph/entities/a.json', content: '{}' }), + ).rejects.toThrow('ENOSPC'); + + expect(exit).toHaveBeenCalledWith(OUT_OF_DISK_EXIT_CODE); +}); + +test('exits with the out-of-disk code when an index symlink runs out of space', async () => { + const exit = mockProcessExit(); + jest.spyOn(fs, 'symlink').mockRejectedValue(outOfDiskError()); + + await expect( + symlink({ sourcePath: 'graph/a.json', destinationPath: 'index/a.json' }), + ).rejects.toThrow('ENOSPC'); + + expect(exit).toHaveBeenCalledWith(OUT_OF_DISK_EXIT_CODE); +}); + +test('exits with the out-of-disk code when graph objects are flushed as JSON', async () => { + const exit = mockProcessExit(); + jest.spyOn(fs, 'writeFile').mockRejectedValue(outOfDiskError()); + + await expect( + writeJsonToPath({ path: 'graph/entities/a.json', data: { entities: [] } }), + ).rejects.toThrow('ENOSPC'); + + expect(exit).toHaveBeenCalledWith(OUT_OF_DISK_EXIT_CODE); +}); + +test('reports why the process died before exiting', async () => { + mockProcessExit(); + jest.spyOn(fs, 'writeFile').mockRejectedValue(outOfDiskError()); + + await expect( + writeFileToPath({ path: 'graph/entities/a.json', content: '{}' }), + ).rejects.toThrow('ENOSPC'); + + // Written to stderr synchronously: `process.exit` does not flush pending + // async writes, so an ordinary logger call would never make it to CloudWatch. + expect(mockedFs.writeSync).toHaveBeenCalledTimes(1); + const [fd, line] = mockedFs.writeSync.mock.calls[0]; + expect(fd).toBe(2); + expect(JSON.parse(line as string)).toMatchObject({ + exitCode: OUT_OF_DISK_EXIT_CODE, + path: expect.stringContaining('a.json'), + }); +}); + +test('still exits when the diagnostic itself cannot be written', async () => { + const exit = mockProcessExit(); + mockedFs.writeSync.mockImplementation(() => { + throw new Error('EBADF: bad file descriptor'); + }); + jest.spyOn(fs, 'writeFile').mockRejectedValue(outOfDiskError()); + + await expect( + writeFileToPath({ path: 'graph/entities/a.json', content: '{}' }), + ).rejects.toThrow('ENOSPC'); + + expect(exit).toHaveBeenCalledWith(OUT_OF_DISK_EXIT_CODE); +}); + +test('leaves every other write failure to normal error handling', async () => { + const exit = mockProcessExit(); + const permissionDenied = Object.assign(new Error('EACCES'), { + code: 'EACCES', + }); + jest.spyOn(fs, 'writeFile').mockRejectedValue(permissionDenied); + + await expect( + writeFileToPath({ path: 'graph/entities/a.json', content: '{}' }), + ).rejects.toThrow('EACCES'); + + expect(exit).not.toHaveBeenCalled(); +}); diff --git a/packages/integration-sdk-runtime/src/fileSystem.ts b/packages/integration-sdk-runtime/src/fileSystem.ts index 668ffaffb..2e9fa8d50 100644 --- a/packages/integration-sdk-runtime/src/fileSystem.ts +++ b/packages/integration-sdk-runtime/src/fileSystem.ts @@ -4,7 +4,7 @@ * This module exports utilities for writing data * relative to the .j1-integration root storage directoryPath. */ -import { promises as fs } from 'fs'; +import { promises as fs, writeSync } from 'fs'; import path from 'path'; import rimraf from 'rimraf'; @@ -20,6 +20,62 @@ const brotliDecompress = promisify(zlib.brotliDecompress); export const DEFAULT_STORAGE_DIRECTORY_NAME = '.j1-integration'; +/** + * Exit code claimed for "the volume backing the storage directory is full". + * + * The managed ECS state machine treats it as a request to retry the task on a + * larger disk (see `handleTaskFailure` in jupiter-integration-service). Any + * other non-zero code is classified as a permanent failure, so this has to stay + * in sync with that handler. + */ +export const OUT_OF_DISK_EXIT_CODE = 77; + +function isOutOfDiskError(error: unknown): boolean { + return (error as NodeJS.ErrnoException | undefined)?.code === 'ENOSPC'; +} + +/** + * Kills the process when a write fails because the disk is full. + * + * Every graph object the integration collects lands on disk before it is + * uploaded, so once the volume is full nothing downstream can succeed. Letting + * the ENOSPC propagate as a normal error is actively harmful: the step executor + * catches it, marks that one step failed, and carries on to the next step, + * which fails the same way. The job then finishes "with errors" and exits 0 — + * a clean task exit that the state machine never catches, so the disk is never + * scaled and the retry never happens. Publishing a partial graph also looks to + * the customer like their data disappeared rather than like a failed run. + * + * Writes the diagnostic synchronously because `process.exit` does not flush + * pending async stdout writes, and this line is the only evidence of why the + * task died. + */ +function exitIfOutOfDisk(error: unknown, fullPath: string): void { + if (!isOutOfDiskError(error)) return; + + try { + writeSync( + 2, + // Shaped as a bunyan record so it lands in the log pipeline alongside + // everything else the integration emitted, rather than as loose text. + `${JSON.stringify({ + v: 0, + name: 'integration-sdk-runtime', + level: 60, + msg: 'Out of disk space while writing collected data. Exiting so the task can be retried with a larger volume.', + path: fullPath, + storageDirectory: getRootStorageDirectory(), + exitCode: OUT_OF_DISK_EXIT_CODE, + time: new Date().toISOString(), + })}\n`, + ); + } catch { + // Nothing useful to do if even stderr is unavailable; still exit below. + } + + process.exit(OUT_OF_DISK_EXIT_CODE); +} + export function getRootStorageDirectory() { return ( process.env.JUPITERONE_INTEGRATION_STORAGE_DIRECTORY || @@ -81,12 +137,19 @@ export async function writeFileToPath({ const directory = getRootStorageDirectory(); const fullPath = path.resolve(directory, relativePath); - await ensurePathCanBeWrittenTo(fullPath); + try { + await ensurePathCanBeWrittenTo(fullPath); - if (isCompressionEnabled()) { - await fs.writeFile(fullPath, await brotliCompress(content), 'utf8'); - } else { - await fs.writeFile(fullPath, content, 'utf8'); + if (isCompressionEnabled()) { + await fs.writeFile(fullPath, await brotliCompress(content), 'utf8'); + } else { + await fs.writeFile(fullPath, content, 'utf8'); + } + } catch (error) { + // Every write of collected data funnels through here, so this is the one + // place that has to notice the volume filling up. + exitIfOutOfDisk(error, fullPath); + throw error; } } @@ -129,15 +192,22 @@ export async function symlink({ sourcePath, destinationPath }: SymlinkInput) { const fullSourcePath = path.resolve(directory, sourcePath); const fullDestinationPath = path.resolve(directory, destinationPath); - await ensurePathCanBeWrittenTo(fullDestinationPath); - // On Windows, we need to perform hardlinks for files - if ( - process.platform === 'win32' && - (await fs.lstat(fullSourcePath)).isFile() - ) { - await fs.link(fullSourcePath, fullDestinationPath); - } else { - await fs.symlink(fullSourcePath, fullDestinationPath, 'junction'); + try { + await ensurePathCanBeWrittenTo(fullDestinationPath); + // On Windows, we need to perform hardlinks for files + if ( + process.platform === 'win32' && + (await fs.lstat(fullSourcePath)).isFile() + ) { + await fs.link(fullSourcePath, fullDestinationPath); + } else { + await fs.symlink(fullSourcePath, fullDestinationPath, 'junction'); + } + } catch (error) { + // The index symlinks are written per flush alongside the graph object + // files, so they hit ENOSPC on the same boundary. + exitIfOutOfDisk(error, fullDestinationPath); + throw error; } }