fix(runtime): exit 77 when collected data cannot be written for lack of disk

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();
+});

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;
   }
 }