Merge pull request #939 from tdurieux/fix/cached-files-without-github

perf: serve cached OAuth snapshots without GitHub calls

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,57 @@ 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); }
+            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();
+          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<string, string | string[] | undefined> }) => {
+          const forwardHeaders = (upstream: { headers: Record<string, string | string[] | undefined> }) => {
             headersDone();
             if (res.headersSent) return;
             for (const header of ["content-encoding", "content-length", "vary", "accept-ranges"]) {
@@ -503,7 +525,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) => {

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<boolean> {
+  // 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;
+}

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

@@ -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
+      | string,
+    cacheOnly = false
   ) {
     const meta = await fileMeta();
     const expected: { sha: string; size?: number; upstream?: UpstreamFile } =
@@ -396,6 +397,28 @@ 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 });
+      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.
+      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 });

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<string, unknown>, headers: Record<string, string>, signal?: AbortSignal) {
+  const stream = got.stream(url, { method: "POST", json: { ...json, token: "", cacheOnly: true },
+    headers, timeout: { response: 15000 }, decompress: false });
+  const responseReady = new Promise<Response>((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;
+  }
+}

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

test/cached-file-probe.test.js

@@ -0,0 +1,220 @@
+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('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;
+    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 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$/, '');
+    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/);
+  });
+});
+
+
+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);
+  });
+});