fix: preserve cache fallback and stream failures during access checks

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

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.

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