From ee107fa6eb8d840141ef3416cff03468777fe803 Mon Sep 17 00:00:00 2001 From: Jean Schmidt Date: Fri, 12 Jun 2026 07:48:28 -0700 Subject: [PATCH] Drain tar writer before destroy in execCpFromPod MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Await `stream.finished(writerStream)` on the success path before destroying, bounded by `AbortSignal.timeout`, so tar-fs finishes writing extracted files to disk - Keep the failure path's immediate destroy + rethrow so the existing 30-attempt retry loop is unchanged - Add `ENV_TAR_DRAIN_TIMEOUT_MS` constant, `DEFAULT_TAR_DRAIN_TIMEOUT_MS = 60000`, and `tarDrainTimeoutMs()` accessor following the existing `useKubeScheduler` pattern - Document `ACTIONS_RUNNER_TAR_DRAIN_TIMEOUT_MS` in `packages/k8s/README.md` under a new Configuration section - Add accessor unit tests (default, valid override, NaN, zero/negative, empty) and three integration tests for `execCpFromPod` covering late-end success, drain timeout, and failure path - Add `destroy: jest.fn()` to the tar.extract mock in `error-serialization-test.ts` so the new unconditional destroy does not crash an unrelated test Notes The kc exec status callback fires when the remote tar process exits, but the WebSocket may still have tar bytes in flight that the local tar-fs extractor has not yet flushed to disk. Destroying the writer at that point silently truncates the extracted workspace. The success path now awaits `stream.promises.finished(writerStream, { signal: AbortSignal.timeout(timeoutMs) })`, so the destroy happens only after tar-fs has closed its file descriptors. The timeout bounds a malformed-stream hang; if it fires, a `core.warning` is emitted and the (possibly incomplete) extract is finalized rather than retrying. The failure path is intentionally not drained — awaiting on a writer whose upstream kc channel reported Failure can hang against a half-open WebSocket. --- packages/k8s/README.md | 8 + packages/k8s/src/k8s/index.ts | 71 +++-- packages/k8s/src/k8s/utils.ts | 15 + .../k8s/tests/error-serialization-test.ts | 3 +- .../k8s/tests/exec-cp-from-pod-drain-test.ts | 260 ++++++++++++++++++ packages/k8s/tests/k8s-utils-test.ts | 43 ++- 6 files changed, 376 insertions(+), 24 deletions(-) create mode 100644 packages/k8s/tests/exec-cp-from-pod-drain-test.ts diff --git a/packages/k8s/README.md b/packages/k8s/README.md index ecc893b8..0938774d 100644 --- a/packages/k8s/README.md +++ b/packages/k8s/README.md @@ -42,3 +42,11 @@ rules: - Docker [create options](https://docs.github.com/en/actions/using-workflows/workflow-syntax-for-github-actions#jobsjob_idcontaineroptions) are not supported - Container actions will have to specify the entrypoint, since the default entrypoint will be overridden to run the commands from the workflow. - Container actions need to have the following binaries in their container image: `sh`, `env`, `tail`. + +## Configuration + +These environment variables are read on the runner container of the runner pod. All knobs are optional; defaults preserve the historical behavior of the hooks. + +| Variable | Default | Purpose | +|---|---|---| +| `ACTIONS_RUNNER_TAR_DRAIN_TIMEOUT_MS` | `60000` | Upper bound (milliseconds) on how long `execCpFromPod` waits for the tar extraction stream to drain after the underlying `kubectl exec` channel reports success. The wait is necessary because the Kubernetes exec status callback fires when the remote tar process exits, but the WebSocket may still have tar bytes in flight that the local `tar-fs` extractor has not yet written to disk; returning early truncates the extracted workspace. Tune up (e.g. `90000`) for very large workspaces over slow links; if the timeout fires, a warning is logged and the (possibly incomplete) extract is finalized. Invalid or non-positive values fall back to the default. | diff --git a/packages/k8s/src/k8s/index.ts b/packages/k8s/src/k8s/index.ts index beaee808..4d90f4a7 100644 --- a/packages/k8s/src/k8s/index.ts +++ b/packages/k8s/src/k8s/index.ts @@ -20,6 +20,7 @@ import { fixArgs, listDirAllCommand, sleep, + tarDrainTimeoutMs, EXTERNALS_VOLUME_NAME, GITHUB_VOLUME_NAME, WORK_VOLUME @@ -585,30 +586,56 @@ export async function execCpFromPod( const writerStream = tar.extract(parentRunnerPath) const errStream = new WritableStreamBuffer() - await new Promise((resolve, reject) => { - exec - .exec( - namespace(), - podName, - JOB_CONTAINER_NAME, - command, - writerStream, - errStream, - null, - false, - async status => { - if (errStream.size()) { - reject( - new Error( - `Error from cpFromPod - details: \n ${errStream.getContentsAsString()}` + try { + await new Promise((resolve, reject) => { + exec + .exec( + namespace(), + podName, + JOB_CONTAINER_NAME, + command, + writerStream, + errStream, + null, + false, + async status => { + if (errStream.size()) { + reject( + new Error( + `Error from cpFromPod - details: \n ${errStream.getContentsAsString()}` + ) ) - ) + } + resolve(status) } - resolve(status) - } - ) - .catch(e => reject(e)) - }) + ) + .catch(e => reject(e)) + }) + } catch (error) { + // Failure path: destroy immediately. Awaiting drain on a stream that + // kc reported Failure on could hang against a half-open WebSocket. + writerStream.destroy() + throw error + } + + // Success path: kc reported the exec channel closed cleanly, but the + // tar bytes from the WebSocket may still be in flight to tar-fs's + // extractor. Returning before the extractor finishes truncates the + // workspace silently. Bound the wait so a malformed stream cannot + // hang the hook indefinitely. + const timeoutMs = tarDrainTimeoutMs() + try { + await stream.promises.finished(writerStream, { + signal: AbortSignal.timeout(timeoutMs) + }) + } catch (err) { + core.warning( + `[execCpFromPod] tar drain did not complete within ${timeoutMs}ms; ` + + `the extracted workspace may be incomplete: ${formatError(err)}` + ) + } finally { + writerStream.destroy() + } break } catch (error) { core.debug(`Attempt ${attempt + 1} failed: ${error}`) diff --git a/packages/k8s/src/k8s/utils.ts b/packages/k8s/src/k8s/utils.ts index 20da57a3..e4358745 100644 --- a/packages/k8s/src/k8s/utils.ts +++ b/packages/k8s/src/k8s/utils.ts @@ -12,6 +12,9 @@ export const DEFAULT_CONTAINER_ENTRY_POINT = 'tail' export const ENV_HOOK_TEMPLATE_PATH = 'ACTIONS_RUNNER_CONTAINER_HOOK_TEMPLATE' export const ENV_USE_KUBE_SCHEDULER = 'ACTIONS_RUNNER_USE_KUBE_SCHEDULER' +export const ENV_TAR_DRAIN_TIMEOUT_MS = 'ACTIONS_RUNNER_TAR_DRAIN_TIMEOUT_MS' + +export const DEFAULT_TAR_DRAIN_TIMEOUT_MS = 60000 export const EXTERNALS_VOLUME_NAME = 'externals' export const GITHUB_VOLUME_NAME = 'github' @@ -269,6 +272,18 @@ export function useKubeScheduler(): boolean { return process.env[ENV_USE_KUBE_SCHEDULER] === 'true' } +export function tarDrainTimeoutMs(): number { + const raw = process.env[ENV_TAR_DRAIN_TIMEOUT_MS] + if (raw === undefined || raw === '') { + return DEFAULT_TAR_DRAIN_TIMEOUT_MS + } + const parsed = parseInt(raw, 10) + if (Number.isNaN(parsed) || parsed <= 0) { + return DEFAULT_TAR_DRAIN_TIMEOUT_MS + } + return parsed +} + export enum PodPhase { PENDING = 'Pending', RUNNING = 'Running', diff --git a/packages/k8s/tests/error-serialization-test.ts b/packages/k8s/tests/error-serialization-test.ts index 47cd4492..9e507f46 100644 --- a/packages/k8s/tests/error-serialization-test.ts +++ b/packages/k8s/tests/error-serialization-test.ts @@ -34,7 +34,8 @@ jest.mock('tar-fs', () => ({ pack: jest.fn().mockReturnValue({ pipe: jest.fn() }), extract: jest.fn().mockReturnValue({ on: jest.fn(), - pipe: jest.fn() + pipe: jest.fn(), + destroy: jest.fn() }) }, __esModule: true diff --git a/packages/k8s/tests/exec-cp-from-pod-drain-test.ts b/packages/k8s/tests/exec-cp-from-pod-drain-test.ts new file mode 100644 index 00000000..5ccbfc5c --- /dev/null +++ b/packages/k8s/tests/exec-cp-from-pod-drain-test.ts @@ -0,0 +1,260 @@ +import { Writable } from 'stream' + +const mockExec = jest.fn() +const mockWarning = jest.fn() +const mockDebug = jest.fn() + +jest.mock('@kubernetes/client-node', () => { + return { + KubeConfig: jest.fn().mockImplementation(() => ({ + loadFromDefault: jest.fn(), + makeApiClient: jest.fn().mockReturnValue({}), + getContexts: jest.fn().mockReturnValue([{ namespace: 'test-namespace' }]) + })), + Exec: jest.fn().mockImplementation(() => ({ exec: mockExec })), + // eslint-disable-next-line @typescript-eslint/no-extraneous-class + CoreV1Api: class CoreV1Api {}, + // eslint-disable-next-line @typescript-eslint/no-extraneous-class + BatchV1Api: class BatchV1Api {}, + // eslint-disable-next-line @typescript-eslint/no-extraneous-class + AuthorizationV1Api: class AuthorizationV1Api {}, + Log: jest.fn() + } +}) + +// Single shared "current writer" the prod code receives from tar.extract. +// Each test replaces it before invoking execCpFromPod. +let currentWriter: Writable | null = null + +jest.mock('tar-fs', () => ({ + default: { + pack: jest.fn().mockReturnValue({ pipe: jest.fn() }), + extract: jest.fn().mockImplementation(() => { + if (!currentWriter) { + throw new Error('test forgot to assign currentWriter before extract()') + } + return currentWriter + }) + }, + __esModule: true +})) + +jest.mock('../src/k8s/utils', () => { + const actual = jest.requireActual('../src/k8s/utils') + return { + ...actual, + sleep: jest.fn().mockResolvedValue(undefined) + } +}) + +jest.mock('@actions/core', () => ({ + info: jest.fn(), + warning: (...args: unknown[]) => mockWarning(...args), + debug: (...args: unknown[]) => mockDebug(...args), + error: jest.fn() +})) + +import { execCpFromPod } from '../src/k8s' +import { ENV_TAR_DRAIN_TIMEOUT_MS } from '../src/k8s/utils' + +// Drive the kc Exec callback once: optionally write some stderr bytes, then +// invoke the status callback so the prod settle promise resolves/rejects. +function mockExecOnceWithStatus( + statusValue: { status: string | undefined }, + opts: { + stderrWrite?: Buffer + // If provided, called after statusCb so the test can simulate "data is + // still landing on writerStream after kc reported Success". Awaited so + // the writer can finish (or hang) before the test continues. + afterStatus?: () => Promise + } = {} +): void { + mockExec.mockImplementationOnce( + async ( + _ns: string, + _pod: string, + _container: string, + _command: string[], + _stdout: NodeJS.WritableStream | null, + stderr: NodeJS.WritableStream, + _stdin: NodeJS.ReadableStream | null, + _tty: boolean, + statusCb: (s: { status: string | undefined }) => void + ) => { + if (opts.stderrWrite) { + ;(stderr as unknown as { write: (b: Buffer) => void }).write( + opts.stderrWrite + ) + } + statusCb(statusValue) + if (opts.afterStatus) { + await opts.afterStatus() + } + return { on: jest.fn() } + } + ) +} + +// Stub out the hash-verification loop's exec calls so the test only exercises +// the tar settle + drain path. The verification loop calls exec via +// execCalculateOutputHashSorted (a fresh `new k8s.Exec(kc)` per attempt). +// All those calls land on the same mockExec; queueing the same response for +// each verification attempt keeps the loop deterministic. +function stubVerificationLoop(): void { + for (let i = 0; i < 32; i++) { + mockExec.mockImplementationOnce( + async ( + _ns: string, + _pod: string, + _container: string, + _command: string[], + stdout: NodeJS.WritableStream | null, + _stderr: NodeJS.WritableStream, + _stdin: NodeJS.ReadableStream | null, + _tty: boolean, + statusCb: (s: { status: string }) => void + ) => { + if (stdout) { + ;(stdout as unknown as { write: (b: Buffer) => void }).write( + Buffer.from('') + ) + } + statusCb({ status: 'Success' }) + return { on: jest.fn() } + } + ) + } +} + +describe('execCpFromPod tar drain', () => { + const originalDrainEnv = process.env[ENV_TAR_DRAIN_TIMEOUT_MS] + + beforeEach(() => { + jest.clearAllMocks() + mockExec.mockReset() + mockWarning.mockReset() + mockDebug.mockReset() + currentWriter = null + process.env['ACTIONS_RUNNER_KUBERNETES_NAMESPACE'] = 'test-namespace' + }) + + afterEach(() => { + delete process.env['ACTIONS_RUNNER_KUBERNETES_NAMESPACE'] + if (originalDrainEnv === undefined) { + delete process.env[ENV_TAR_DRAIN_TIMEOUT_MS] + } else { + process.env[ENV_TAR_DRAIN_TIMEOUT_MS] = originalDrainEnv + } + }) + + it('awaits writer drain on Success: late writer end is honored, no warning', async () => { + // Use a plain Writable so stream.finished resolves on `end()` without + // needing a downstream reader. A PassThrough would buffer forever + // because nothing pipes its readable side. + const writer = new Writable({ + write(_chunk, _enc, cb) { + cb() + } + }) + currentWriter = writer + + process.env[ENV_TAR_DRAIN_TIMEOUT_MS] = '5000' + + mockExecOnceWithStatus( + { status: 'Success' }, + { + // Simulate tar bytes still landing AFTER kc reports Success, then a + // graceful end. The drain await must observe this end. + afterStatus: async () => { + await new Promise(r => setTimeout(r, 30)) + writer.write(Buffer.from('late tar bytes')) + writer.end() + } + } + ) + // Verification-loop stubs MUST be queued AFTER the tar exec stub — + // mockImplementationOnce is FIFO. + stubVerificationLoop() + + await expect( + execCpFromPod('my-pod', '/workspace/output', '/tmp/dst') + ).resolves.toBeUndefined() + + // Drain completed cleanly — the timeout warning MUST NOT fire. + const warnMessages = mockWarning.mock.calls.map(c => String(c[0])) + expect( + warnMessages.some(m => m.includes('tar drain did not complete')) + ).toBe(false) + }) + + it('Success + writer that never ends: timeout fires, warning emitted, writer destroyed, promise resolves', async () => { + const writer = new Writable({ + write(_chunk, _enc, cb) { + cb() + } + }) + currentWriter = writer + + // Small timeout so the test does not slow the suite. The drain bound + // must be measured in real time (AbortSignal.timeout), so a tiny value + // is the only way to keep this test fast. + process.env[ENV_TAR_DRAIN_TIMEOUT_MS] = '50' + + const destroySpy = jest.spyOn(writer, 'destroy') + + mockExecOnceWithStatus({ status: 'Success' }) + // Deliberately do NOT call writer.end(). The drain await will trip the + // 50ms AbortSignal timeout. + // Verification stubs come AFTER the tar exec stub (FIFO). + stubVerificationLoop() + + await expect( + execCpFromPod('my-pod', '/workspace/output', '/tmp/dst') + ).resolves.toBeUndefined() + + const warnMessages = mockWarning.mock.calls.map(c => String(c[0])) + expect( + warnMessages.some(m => + m.includes('tar drain did not complete within 50ms') + ) + ).toBe(true) + expect(destroySpy).toHaveBeenCalled() + }) + + it('Failure status: writer is destroyed immediately, no drain await', async () => { + const writer = new Writable({ + write(_chunk, _enc, cb) { + cb() + } + }) + currentWriter = writer + + process.env[ENV_TAR_DRAIN_TIMEOUT_MS] = '60000' + const destroySpy = jest.spyOn(writer, 'destroy') + + // Upstream's settle logic only rejects when errStream is non-empty; a + // bare Failure status with no stderr resolves. Force the reject path + // by writing stderr on every attempt. execCpFromPod retries 30 times, + // each attempt creating a fresh writer via tar.extract — currentWriter + // is reused across attempts which is fine for the destroy assertion. + for (let i = 0; i < 30; i++) { + mockExecOnceWithStatus( + { status: 'Failure' }, + { stderrWrite: Buffer.from('tar: cannot open: Permission denied') } + ) + } + + await expect( + execCpFromPod('my-pod', '/workspace/output', '/tmp/dst') + ).rejects.toThrow() + + // Writer was destroyed on the failure path — at least once per attempt. + expect(destroySpy).toHaveBeenCalled() + // Critically: the drain timeout warning must NOT fire on the failure + // path. Failure path skips the drain entirely. + const warnMessages = mockWarning.mock.calls.map(c => String(c[0])) + expect( + warnMessages.some(m => m.includes('tar drain did not complete')) + ).toBe(false) + }) +}) diff --git a/packages/k8s/tests/k8s-utils-test.ts b/packages/k8s/tests/k8s-utils-test.ts index bfb6c453..942eee20 100644 --- a/packages/k8s/tests/k8s-utils-test.ts +++ b/packages/k8s/tests/k8s-utils-test.ts @@ -6,7 +6,9 @@ import { mergePodSpecWithOptions, mergeContainerWithOptions, readExtensionFromFile, - ENV_HOOK_TEMPLATE_PATH + tarDrainTimeoutMs, + ENV_HOOK_TEMPLATE_PATH, + ENV_TAR_DRAIN_TIMEOUT_MS } from '../src/k8s/utils' import * as k8s from '@kubernetes/client-node' import { TestHelper } from './test-setup' @@ -406,4 +408,43 @@ spec: expect(base).toStrictEqual(expected) }) + + describe('tarDrainTimeoutMs', () => { + const original = process.env[ENV_TAR_DRAIN_TIMEOUT_MS] + + afterEach(() => { + if (original === undefined) { + delete process.env[ENV_TAR_DRAIN_TIMEOUT_MS] + } else { + process.env[ENV_TAR_DRAIN_TIMEOUT_MS] = original + } + }) + + it('returns the 60s default when the env var is unset', () => { + delete process.env[ENV_TAR_DRAIN_TIMEOUT_MS] + expect(tarDrainTimeoutMs()).toBe(60000) + }) + + it('returns the parsed integer when the env var is a valid positive number', () => { + process.env[ENV_TAR_DRAIN_TIMEOUT_MS] = '12345' + expect(tarDrainTimeoutMs()).toBe(12345) + }) + + it('falls back to the default when the env var is not a number', () => { + process.env[ENV_TAR_DRAIN_TIMEOUT_MS] = 'not-a-number' + expect(tarDrainTimeoutMs()).toBe(60000) + }) + + it('falls back to the default when the env var is zero or negative', () => { + process.env[ENV_TAR_DRAIN_TIMEOUT_MS] = '0' + expect(tarDrainTimeoutMs()).toBe(60000) + process.env[ENV_TAR_DRAIN_TIMEOUT_MS] = '-100' + expect(tarDrainTimeoutMs()).toBe(60000) + }) + + it('falls back to the default when the env var is an empty string', () => { + process.env[ENV_TAR_DRAIN_TIMEOUT_MS] = '' + expect(tarDrainTimeoutMs()).toBe(60000) + }) + }) })