From 4fc3c2e042d4fe2c02b7f720f944f2f9df224933 Mon Sep 17 00:00:00 2001 From: Codex Date: Tue, 10 Nov 2026 12:03:52 +0000 Subject: [PATCH 1/3] perf: serve cached OAuth snapshots without GitHub calls --- src/core/AnonymizedFile.ts | 69 +++++++----- src/core/cached-repository-access.ts | 36 ++++++ src/core/source/GitHubStream.ts | 20 +++- src/core/streamer-cache-probe.ts | 24 ++++ src/streamer/route.ts | 9 +- test/cached-file-probe.test.js | 162 +++++++++++++++++++++++++++ 6 files changed, 291 insertions(+), 29 deletions(-) create mode 100644 src/core/cached-repository-access.ts create mode 100644 src/core/streamer-cache-probe.ts create mode 100644 test/cached-file-probe.test.js diff --git a/src/core/AnonymizedFile.ts b/src/core/AnonymizedFile.ts index 2a7e81f..3f6873b 100644 --- a/src/core/AnonymizedFile.ts +++ b/src/core/AnonymizedFile.ts @@ -1,3 +1,5 @@ +import { canReadCachedRepository } from "./cached-repository-access"; +import { probeStreamerCache } from "./streamer-cache-probe"; import { notifyOwnerAccessProblem } from "./owner-notifications"; import { sendSharedContent } from "./cached-response"; import { streamResponse } from "./response-stream"; @@ -458,37 +460,48 @@ export default class AnonymizedFile { try { if (config.STREAMER_ENTRYPOINT) { // use the streamer service - const [sha, size, token] = await Promise.all([ - this.sha(), - this.size(), - this.repository.getToken(), - ]); + const [sha, size] = await Promise.all([this.sha(), this.size()]); + const payload = { + sha, + size, + upstream: (await this.getFileInfo()).upstream, + repoFullName: this.repository.model.source.repositoryName, + commit: this.repository.model.source.commit, + cacheGeneration: `${this.repository.model.treeGeneration || "legacy"}:${this.repository.model.anonymizeDate?.toISOString() || ""}`, + cacheRevision: this.repository.model.contentCacheRevision, + branch: this.repository.model.source.branch, + repoId: this.repository.repoId, + filePath: this.filePath, + anonymizerOptions: anonymizer.opt, + }; + const url = join(config.STREAMER_ENTRYPOINT, "api"); + const headers = { ...requestHeaders(), "accept-encoding": res.req?.get?.("accept-encoding") || "identity" }; const headersDone = startStage("streamer_headers"); - const resStream = got - .stream(join(config.STREAMER_ENTRYPOINT, "api"), { - method: "POST", - headers: { ...requestHeaders(), "accept-encoding": res.req?.get?.("accept-encoding") || "identity" }, - decompress: false, - json: { - sha, - size, - upstream: (await this.getFileInfo()).upstream, - token: await githubTokenForStreamer(token, this.repository.model.source.repositoryName), - repoFullName: this.repository.model.source.repositoryName, - commit: this.repository.model.source.commit, - cacheGeneration: `${this.repository.model.treeGeneration || "legacy"}:${this.repository.model.anonymizeDate?.toISOString() || ""}`, - cacheRevision: this.repository.model.contentCacheRevision, - branch: this.repository.model.source.branch, - repoId: this.repository.repoId, - filePath: this.filePath, - anonymizerOptions: anonymizer.opt, - }, - }); + const probe = async () => { + if (!config.SHARED_CACHE_ENABLED || !await canReadCachedRepository(this.repository)) return undefined; + const cancellation = new AbortController(); + const cancel = () => cancellation.abort(); + res.once("close", cancel); + if (res.destroyed || res.writableEnded) cancel(); + try { return await probeStreamerCache(url, payload, headers, cancellation.signal); } + finally { res.removeListener("close", cancel); } + }; + const cached = await probe(); + if (cached) { + try { + if (!await canReadCachedRepository(this.repository)) throw new AnonymousError("connection_changed", { httpStatus: 409 }); + } catch (error) { cached.stream.destroy(); throw error; } + } + if (res.destroyed || res.writableEnded) { cached?.stream.destroy(); headersDone(); resolve(); return; } + const resStream = cached?.stream || got.stream(url, { + method: "POST", headers, decompress: false, + json: { ...payload, token: await githubTokenForStreamer(await this.repository.getToken(), this.repository.model.source.repositoryName) }, + }); // Forward Content-Type from the streamer's upstream response. // got.stream(...).pipe(res) forwards body bytes only — without // this, the parent response has no Content-Type and the browser // guesses (text renders as download, images as octet-stream). - resStream.on("response", (upstream: { headers: Record }) => { + const forwardHeaders = (upstream: { headers: Record }) => { headersDone(); if (res.headersSent) return; for (const header of ["content-encoding", "content-length", "vary", "accept-ranges"]) { @@ -503,7 +516,9 @@ export default class AnonymizedFile { if (fallback) res.contentType(fallback); else if (isTextFile(this.anonymizedPath)) res.contentType("text/plain"); } - }); + }; + if (cached) forwardHeaders(cached.response); + else resStream.on("response", forwardHeaders); resStream.once("error", () => headersDone(true)); resStream.once("close", () => headersDone()); await streamResponse(resStream, res, (err: Error) => { diff --git a/src/core/cached-repository-access.ts b/src/core/cached-repository-access.ts new file mode 100644 index 0000000..6e38649 --- /dev/null +++ b/src/core/cached-repository-access.ts @@ -0,0 +1,36 @@ +import Repository from "./Repository"; +import AnonymousError from "./AnonymousError"; +import { isConnected } from "../server/database"; +import RepositoryModel from "./model/anonymizedRepositories/anonymizedRepositories.model"; +import UserModel from "./model/users/users.model"; +import CredentialModel from "./model/credentials/credentials.model"; +import config from "../config"; +import { APP_PROVIDER, appError } from "./github-app"; + +/** Local access to an already published snapshot, not permission to fetch upstream. */ +export async function canReadCachedRepository(repository: Repository): Promise { + // Unpersisted/test objects must use the existing authorization path. + if (!isConnected || repository.model.isNew) return false; + await repository.check(); + const current = await RepositoryModel.findById(repository.model._id).select("owner githubAccess").lean(); + if (!current || String(current.owner) !== repository.owner.id || + current.githubAccess?.revision !== repository.model.githubAccess?.revision) { + throw new AnonymousError("connection_changed", { httpStatus: 409 }); + } + const user = await UserModel.findById(repository.owner.id).select("status").lean(); + if (!user || user.status === "removed" || user.status === "banned") { + throw new AnonymousError("not_authorized", { httpStatus: 403 }); + } + const binding = current.githubAccess; + const grant = await CredentialModel.findOne({ ownerId: repository.owner.id, + provider: binding?.kind === "github-app" ? APP_PROVIDER : "github" }).select("revoked").lean(); + if (grant?.revoked) throw new AnonymousError("not_authorized", { httpStatus: 403 }); + if (binding?.kind === "github-app") { + if (!config.GITHUB_APP_ENABLED || !grant) throw appError(); + // Public visibility and user membership changes are not fully covered by + // local webhook state. Keep remote authorization for these bindings until + // a durable per-repository permission reconciliation mechanism exists. + return false; + } + return true; +} diff --git a/src/core/source/GitHubStream.ts b/src/core/source/GitHubStream.ts index f9de7ae..caba409 100644 --- a/src/core/source/GitHubStream.ts +++ b/src/core/source/GitHubStream.ts @@ -337,7 +337,8 @@ export default class GitHubStream extends GitHubBase { | Promise<{ sha: string; size?: number; upstream?: UpstreamFile }> | { sha: string; size?: number; upstream?: UpstreamFile } | Promise - | string + | string, + cacheOnly = false ) { const meta = await fileMeta(); const expected: { sha: string; size?: number; upstream?: UpstreamFile } = @@ -396,6 +397,23 @@ export default class GitHubStream extends GitHubBase { } }; const shared = sharedBlobCache(); + // A cache-only probe must never fill, promote legacy data or contact GitHub. + // The API has checked local access; lifecycle/revision fences still apply here. + if (cacheOnly) { + await assertActive(); + if (expected.size != null && expected.size > config.MAX_FILE_SIZE) throw new AnonymousError("file_too_big", { httpStatus: 413 }); + const hit = /^[a-f0-9]{40}$/.test(expected.sha) ? await shared?.lookup(`git-${expected.sha}`) : undefined; + if (!hit) throw new AnonymousError("cache_miss", { httpStatus: 409 }); + // Resolved LFS payloads may need additional billing admission. Let the + // normal path reserve that expansion before serving it. + if (config.BILLING_ENABLED && hit.cachedBlob!.record.size > (expected.size || 0)) { + hit.destroy(); throw new AnonymousError("cache_miss", { httpStatus: 409 }); + } + try { await assertActive(); } catch (error) { hit.destroy(); throw error; } + observeCache("source_hit", hit.cachedBlob!.record.size); + recordCacheHit("source", true); + return hit; + } if (shared && /^[a-f0-9]{40}$/.test(expected.sha)) { await assertActive(); if (expected.size != null && expected.size > config.MAX_FILE_SIZE) throw new AnonymousError("file_too_big", { httpStatus: 413 }); diff --git a/src/core/streamer-cache-probe.ts b/src/core/streamer-cache-probe.ts new file mode 100644 index 0000000..eb2d1d1 --- /dev/null +++ b/src/core/streamer-cache-probe.ts @@ -0,0 +1,24 @@ +import got, { Response } from "got"; + +/** Wait only for headers; the body remains a backpressured stream. */ +export async function probeStreamerCache(url: string, json: Record, headers: Record, signal?: AbortSignal) { + const stream = got.stream(url, { method: "POST", json: { ...json, token: "", cacheOnly: true }, + headers, timeout: { response: 15000 }, decompress: false }); + const responseReady = new Promise((resolve, reject) => { + stream.once("response", resolve); + stream.once("error", reject); + }); + const abort = () => stream.destroy(new Error("response_cancelled")); + signal?.addEventListener("abort", abort, { once: true }); + stream.once("close", () => signal?.removeEventListener("abort", abort)); + if (signal?.aborted) abort(); + try { + const response = await responseReady; + return { stream, response }; + } catch (error) { + stream.destroy(); + const response = (error as { response?: Response }).response; + if (response?.statusCode === 409 && response.headers["x-anonymous-cache-miss"] === "1") return undefined; + throw error; + } +} diff --git a/src/streamer/route.ts b/src/streamer/route.ts index 3a64bf0..0e075be 100644 --- a/src/streamer/route.ts +++ b/src/streamer/route.ts @@ -91,7 +91,8 @@ router.post("/", async (req: express.Request, res: express.Response) => { const content = await source.getFileContentCache( filePath, repoId, - () => ({ sha: fileSha, size: fileSize, upstream: req.body.upstream }) + () => ({ sha: fileSha, size: fileSize, upstream: req.body.upstream }), + req.body.cacheOnly === true ); if (res.destroyed || res.writableEnded) { content.destroy(); anonymizer.destroy(); return; } const mime = lookup(filePath); @@ -136,6 +137,12 @@ router.post("/", async (req: express.Request, res: express.Response) => { } }); } catch (error) { + anonymizer.destroy(); + if (req.body.cacheOnly === true && error instanceof Error && error.message === "cache_miss") { + res.setHeader("X-Anonymous-Cache-Miss", "1"); + res.status(409).end(); + return; + } handleError(error, res); } }); diff --git a/test/cached-file-probe.test.js b/test/cached-file-probe.test.js new file mode 100644 index 0000000..0c732ec --- /dev/null +++ b/test/cached-file-probe.test.js @@ -0,0 +1,162 @@ +const { strict: assert } = require('node:assert'); +const { promises: fs } = require('node:fs'); +const { tmpdir } = require('node:os'); +const { join } = require('node:path'); +const { Readable } = require('node:stream'); +const { createHash } = require('node:crypto'); +const express = require('express'); +require('ts-node/register/transpile-only'); +const config = require('../src/config').default; +const { sharedBlobCache, closeBlobCaches } = require('../src/core/blob-cache'); +const { probeStreamerCache } = require('../src/core/streamer-cache-probe'); +const GitHubStream = require('../src/core/source/GitHubStream').default; +const streamer = require('../src/streamer/route').default; +const localAccess = require('../src/core/cached-repository-access'); +const File = require('../src/core/AnonymizedFile').default; +const { AnonymizeTransformer } = require('../src/core/anonymize-utils'); +const got = require('got'); +const body = Buffer.from('Public README with Alice\n'); +const sha = createHash('sha1').update(`blob ${body.length}\0`).update(body).digest('hex'); +const consume = async stream => { let text = ''; for await (const chunk of stream) text += chunk; return text; }; + +describe('cached file streamer probe', function () { + this.timeout(10000); + let root, old, server, url; + beforeEach(async () => { + old = { ...config }; closeBlobCaches(); + root = await fs.mkdtemp(join(tmpdir(), 'cached-probe-')); + Object.assign(config, { STORAGE: 'filesystem', FOLDER: root, SHARED_CACHE_ENABLED: true, + SHARED_CACHE_PATH: join(root, 'cache'), SHARED_CACHE_BYTES: 16 * 1024 ** 2, + SHARED_CACHE_RESULT_BYTES: 8 * 1024 ** 2, SHARED_CACHE_MAX_FILE_BYTES: 1024 ** 2, + SHARED_CACHE_FREE_BYTES: 0, BILLING_ENABLED: false }); + const app = express(); app.use(express.json()); app.use('/api', streamer); + server = await new Promise(resolve => { const s = app.listen(0, '127.0.0.1', () => resolve(s)); }); + url = `http://127.0.0.1:${server.address().port}/api`; + }); + afterEach(async () => { + await new Promise(resolve => server.close(resolve)); closeBlobCaches(); Object.assign(config, old); + await fs.rm(root, { recursive: true, force: true }); + }); + const payload = () => ({ repoFullName: 'owner/private', repoId: 'anon', sha, size: body.length, + filePath: 'README.md', anonymizerOptions: { terms: ['Alice'], filePath: 'README.md' } }); + async function seed() { + const stream = await sharedBlobCache().getOrFill(`git-${sha}`, async () => Readable.from(body), async () => `git-${sha}`, body.length); + await consume(stream); + } + it('returns a cache miss without credentials, downloads, or GitHub calls', async () => { + const source = new GitHubStream({ getToken: () => { throw new Error('credential access'); } }); + source.downloadWithFallback = () => { throw new Error('upstream access'); }; + await assert.rejects(source.getFileContentCache('README.md', 'anon', () => ({ sha, size: body.length }), true), /cache_miss/); + assert.equal(await probeStreamerCache(url, payload(), {}), undefined); + }); + it('serves a hit through the real streamer and anonymizes it without GitHub', async () => { + await seed(); + const fetch = global.fetch; + global.fetch = () => { throw new Error('unexpected GitHub call'); }; + try { + const result = await probeStreamerCache(url, payload(), {}); + assert.equal(result.response.statusCode, 200); + const text = await consume(result.stream); + assert.ok(text.includes('Public README')); + assert.ok(!text.includes('Alice')); + } finally { global.fetch = fetch; } + }); + it('falls back to normal admission for larger billed LFS payloads', async () => { + await seed(); config.BILLING_ENABLED = true; + assert.equal(await probeStreamerCache(url, { ...payload(), size: 1 }, {}), undefined); + }); + it('does not disguise validation failures as cache misses', async () => { + await assert.rejects(probeStreamerCache(url, { ...payload(), filePath: '../secret' }, {}), error => error.response.statusCode === 400); + }); + it('rejects a generation change even when the blob is cached', async () => { + await seed(); + const db = require('../src/server/database'); + const Model = require('../src/core/model/anonymizedRepositories/anonymizedRepositories.model').default; + const connected = db.isConnected, find = Model.findOne; + db.isConnected = true; + Model.findOne = () => ({ select: () => ({ lean: () => ({ exec: async () => ({ status: 'ready', treeGeneration: 'new' }) }) }) }); + try { + const source = new GitHubStream({ cacheGeneration: 'old:', getToken: () => { throw new Error('GitHub called'); } }); + await assert.rejects(source.getFileContentCache('README.md', 'anon', () => ({ sha, size: body.length }), true), /repository_changed/); + } finally { db.isConnected = connected; Model.findOne = find; } + }); + it('cancels a pending probe when the browser disconnects', async () => { + const app = express(); + let closed; + const disconnected = new Promise(resolve => { closed = resolve; }); + app.post('/', (_req, res) => { res.once('close', closed); }); + const pending = await new Promise(resolve => { const s = app.listen(0, '127.0.0.1', () => resolve(s)); }); + const cancel = new AbortController(); + try { + const result = probeStreamerCache(`http://127.0.0.1:${pending.address().port}`, payload(), {}, cancel.signal); + global.setTimeout(() => cancel.abort(), 50); + await assert.rejects(result, /response_cancelled/); + await disconnected; + } finally { await new Promise(resolve => pending.close(resolve)); } + }); + for (const hit of [true, false]) it(`API send ${hit ? 'hit skips' : 'miss performs'} GitHub authorization`, async () => { + if (hit) await seed(); + config.STREAMER_ENTRYPOINT = url.replace(/api$/, ''); + const check = localAccess.canReadCachedRepository; + localAccess.canReadCachedRepository = async () => true; + let tokenCalls = 0; + const app = express(); + app.get('/', async (_req, res) => { + const file = new File({ repository: { + repoId: 'anon', options: { terms: ['Alice'] }, model: { source: { repositoryName: 'owner/private' } }, + getToken: async () => { tokenCalls++; throw new Error('GitHub unavailable'); }, + generateAnonymizeTransformer: filePath => new AnonymizeTransformer({ terms: ['Alice'], filePath }), + }, anonymizedPath: 'README.md' }); + file._file = { name: 'README.md', path: '', sha, size: body.length }; + try { await file.send(res); } catch (error) { res.status(503).send(error.message); } + }); + const api = await new Promise(resolve => { const s = app.listen(0, '127.0.0.1', () => resolve(s)); }); + try { + const response = await got(`http://127.0.0.1:${api.address().port}`, { throwHttpErrors: false, retry: 0 }); + assert.equal(response.statusCode, hit ? 200 : 503); + assert.equal(tokenCalls, hit ? 0 : 1); + if (hit) assert.ok(!response.body.includes('Alice')); + } finally { localAccess.canReadCachedRepository = check; await new Promise(resolve => api.close(resolve)); } + }); +}); + +describe('local cached snapshot access', () => { + const db = require('../src/server/database'); + const RepositoryModel = require('../src/core/model/anonymizedRepositories/anonymizedRepositories.model').default; + const UserModel = require('../src/core/model/users/users.model').default; + const CredentialModel = require('../src/core/model/credentials/credentials.model').default; + let restores, current, user, grant, repository; + const stub = (object, key, value) => { const old = object[key]; restores.push(() => { object[key] = old; }); object[key] = value; }; + beforeEach(() => { + restores = []; + current = { owner: 'owner', githubAccess: { kind: 'oauth', revision: 'r1' } }; + user = { status: 'active' }; grant = { revoked: false }; + repository = { model: { _id: 'repo', githubAccess: { ...current.githubAccess } }, owner: { id: 'owner' }, check: async () => {} }; + stub(db, 'isConnected', true); + stub(RepositoryModel, 'findById', () => ({ select: () => ({ lean: async () => current }) })); + stub(UserModel, 'findById', () => ({ select: () => ({ lean: async () => user }) })); + stub(CredentialModel, 'findOne', () => ({ select: () => ({ lean: async () => grant }) })); + }); + afterEach(() => { for (const restore of restores.reverse()) restore(); }); + it('allows OAuth snapshots without reading or refreshing a token', async () => { + assert.equal(await localAccess.canReadCachedRepository(repository), true); + }); + for (const change of ['owner', 'revision', 'deleted', 'banned', 'removed', 'revoked', 'expired']) { + it(`rejects ${change} before using cached bytes`, async () => { + if (change === 'owner') current.owner = 'other'; + if (change === 'revision') current.githubAccess.revision = 'new'; + if (change === 'deleted') current = null; + if (change === 'banned' || change === 'removed') user.status = change; + if (change === 'revoked') grant.revoked = true; + if (change === 'expired') repository.check = async () => { throw new Error('repository_expired'); }; + await assert.rejects(localAccess.canReadCachedRepository(repository)); + }); + } + it('retains remote authorization for App membership and visibility checks', async () => { + stub(config, 'GITHUB_APP_ENABLED', true); + current.githubAccess.kind = repository.model.githubAccess.kind = 'github-app'; + assert.equal(await localAccess.canReadCachedRepository(repository), false); + grant.revoked = true; + await assert.rejects(localAccess.canReadCachedRepository(repository), /not_authorized/); + }); +}); From 07458f536aa68541d1bea019a2001ee0d97e76d6 Mon Sep 17 00:00:00 2001 From: Codex Date: Tue, 10 Nov 2026 12:51:09 +0000 Subject: [PATCH 2/3] fix: preserve cache fallback and stream failures during access checks --- src/core/response-stream.ts | 5 ++++- src/core/source/GitHubStream.ts | 7 ++++++- test/cached-file-probe.test.js | 31 +++++++++++++++++++++++++++++++ 3 files changed, 41 insertions(+), 2 deletions(-) diff --git a/src/core/response-stream.ts b/src/core/response-stream.ts index af80894..75bb18c 100644 --- a/src/core/response-stream.ts +++ b/src/core/response-stream.ts @@ -19,6 +19,9 @@ export function streamResponse(source: Readable, response: Writable, response.once("finish", done); response.once("error", failed); source.once("error", upstreamFailed); - if (response.destroyed || response.writableEnded) done(); else source.pipe(response); + if (response.destroyed || response.writableEnded) done(); + else if (source.errored) upstreamFailed(source.errored); + else if (source.destroyed || source.readableEnded) upstreamFailed(new Error("upstream_stream_closed")); + else source.pipe(response); }); } diff --git a/src/core/source/GitHubStream.ts b/src/core/source/GitHubStream.ts index caba409..d7cc447 100644 --- a/src/core/source/GitHubStream.ts +++ b/src/core/source/GitHubStream.ts @@ -402,7 +402,12 @@ export default class GitHubStream extends GitHubBase { if (cacheOnly) { await assertActive(); if (expected.size != null && expected.size > config.MAX_FILE_SIZE) throw new AnonymousError("file_too_big", { httpStatus: 413 }); - const hit = /^[a-f0-9]{40}$/.test(expected.sha) ? await shared?.lookup(`git-${expected.sha}`) : undefined; + let hit; + try { hit = /^[a-f0-9]{40}$/.test(expected.sha) ? await shared?.lookup(`git-${expected.sha}`) : undefined; } + catch (error) { + if (!(error instanceof CacheAdmissionError)) throw error; + throw new AnonymousError("cache_miss", { httpStatus: 409 }); + } if (!hit) throw new AnonymousError("cache_miss", { httpStatus: 409 }); // Resolved LFS payloads may need additional billing admission. Let the // normal path reserve that expansion before serving it. diff --git a/test/cached-file-probe.test.js b/test/cached-file-probe.test.js index 0c732ec..093b77d 100644 --- a/test/cached-file-probe.test.js +++ b/test/cached-file-probe.test.js @@ -49,6 +49,17 @@ describe('cached file streamer probe', function () { await assert.rejects(source.getFileContentCache('README.md', 'anon', () => ({ sha, size: body.length }), true), /cache_miss/); assert.equal(await probeStreamerCache(url, payload(), {}), undefined); }); + it('treats a real SQLite write lock as a miss and preserves the cached bytes', async () => { + await seed(); + const { DatabaseSync } = require('node:sqlite'); + const other = new DatabaseSync(join(root, 'cache/source/index.sqlite')); + other.exec('BEGIN IMMEDIATE'); + try { assert.equal(await probeStreamerCache(url, payload(), {}), undefined); } + finally { other.exec('ROLLBACK'); other.close(); } + const result = await probeStreamerCache(url, payload(), {}); + assert.equal(result.response.statusCode, 200); + assert.ok((await consume(result.stream)).includes('Public README')); + }); it('serves a hit through the real streamer and anonymizes it without GitHub', async () => { await seed(); const fetch = global.fetch; @@ -160,3 +171,23 @@ describe('local cached snapshot access', () => { await assert.rejects(localAccess.canReadCachedRepository(repository), /not_authorized/); }); }); + + +describe('response handoff after access recheck', () => { + const { PassThrough } = require('node:stream'); + const { streamResponse } = require('../src/core/response-stream'); + for (const withError of [true, false]) it(`handles an already destroyed upstream ${withError ? 'with its original error' : 'without an error'}`, async () => { + const source = new PassThrough(); + const response = new PassThrough(); + const error = new Error('streamer connection lost during access check'); + source.on('error', () => {}); + const closed = new Promise(resolve => source.once('close', resolve)); + source.destroy(withError ? error : undefined); + await closed; + let received; + await streamResponse(source, response, failure => { received = failure; response.end(); }); + if (withError) assert.equal(received, error); + else assert.equal(received.message, 'upstream_stream_closed'); + assert.equal(response.writableEnded, true); + }); +}); From 0abbe34aed10637381bfcd6a433049405922b964 Mon Sep 17 00:00:00 2001 From: Codex Date: Tue, 10 Nov 2026 12:55:55 +0000 Subject: [PATCH 3/3] fix: preserve streamer error semantics for cache probes --- src/core/AnonymizedFile.ts | 9 +++++++++ test/cached-file-probe.test.js | 27 +++++++++++++++++++++++++++ 2 files changed, 36 insertions(+) diff --git a/src/core/AnonymizedFile.ts b/src/core/AnonymizedFile.ts index 3f6873b..0ad2a35 100644 --- a/src/core/AnonymizedFile.ts +++ b/src/core/AnonymizedFile.ts @@ -484,6 +484,15 @@ export default class AnonymizedFile { res.once("close", cancel); if (res.destroyed || res.writableEnded) cancel(); try { return await probeStreamerCache(url, payload, headers, cancellation.signal); } + catch (err) { + headersDone(true); + const { error } = streamerErrorToAnonymous(err as Error & { + response?: { statusCode?: number; body?: unknown }; + }, { repoId: this.repository.repoId, filePath: this.anonymizedPath }); + void notifyOwnerAccessProblem(this.repository.owner.id, error, { kind: "repository", id: String(this.repository.model._id) }); + error.value = this; + throw error; + } finally { res.removeListener("close", cancel); } }; const cached = await probe(); diff --git a/test/cached-file-probe.test.js b/test/cached-file-probe.test.js index 093b77d..99913f7 100644 --- a/test/cached-file-probe.test.js +++ b/test/cached-file-probe.test.js @@ -105,6 +105,33 @@ describe('cached file streamer probe', function () { await disconnected; } finally { await new Promise(resolve => pending.close(resolve)); } }); + for (const status of [0, 409, 500]) it(`normalizes probe failure ${status || 'connection refused'} at the API boundary`, async () => { + const app = express(); + app.post('/api', (_req, res) => res.status(status || 500).json({ error: status === 409 ? 'repository_changed' : 'upstream_error' })); + const upstream = await new Promise(resolve => { const s = app.listen(0, '127.0.0.1', () => resolve(s)); }); + config.STREAMER_ENTRYPOINT = `http://127.0.0.1:${upstream.address().port}/`; + if (!status) await new Promise(resolve => upstream.close(resolve)); + const check = localAccess.canReadCachedRepository; + localAccess.canReadCachedRepository = async () => true; + const response = new (require('node:stream').PassThrough)(); + const file = new File({ repository: { + repoId: 'anon', owner: { id: '507f1f77bcf86cd799439011' }, options: { terms: [] }, + model: { _id: '507f1f77bcf86cd799439012', source: { repositoryName: 'owner/private' } }, + getToken: async () => { throw new Error('unexpected authorization fallback'); }, + generateAnonymizeTransformer: filePath => new AnonymizeTransformer({ terms: [], filePath }), + }, anonymizedPath: 'README.md' }); + file._file = { name: 'README.md', path: '', sha, size: body.length }; + try { + await assert.rejects(file.send(response), error => { + assert.equal(error.httpStatus, status === 409 ? 409 : 502); + assert.equal(error.message, status === 409 ? 'repository_changed' : status ? 'upstream_error' : 'streamer_unreachable'); + return true; + }); + } finally { + localAccess.canReadCachedRepository = check; response.destroy(); + if (status) await new Promise(resolve => upstream.close(resolve)); + } + }); for (const hit of [true, false]) it(`API send ${hit ? 'hit skips' : 'miss performs'} GitHub authorization`, async () => { if (hit) await seed(); config.STREAMER_ENTRYPOINT = url.replace(/api$/, '');