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