feat: add repository affinity peer sync and r2 failover e2e

Author: mizchi

README.md

@@ -177,6 +177,10 @@ optional env:
 - `RELAY_PEERS_JSON` (JSON: peer relay URLs, `RELAY_PEERS` より優先)
 - `RELAY_PEER_AUTH_TOKEN` (peer pull 時の Bearer token)
 - `RELAY_PEER_SYNC_INTERVAL_SEC` (default: `30`)
+- `RELAY_PEER_REPO_FF_COMMIT_WINDOW` (default: `30`, peer 間 repo 判定で見る commit 数)
+- `RELAY_REPO_ID` (明示 repo ID, 例: `bit-vcs/bit`)
+- `RELAY_REPO_ORIGIN_URL` (origin URL の明示上書き)
+- `RELAY_REPO_RECENT_COMMITS` (CSV: 最近 commit hash)
 - `RELAY_CACHE_PROVIDER` (`memory` or `r2`, default: `memory`)
 - `RELAY_CACHE_TTL_SEC` (default: `86400`)
 - `RELAY_CACHE_MAX_BYTES` (任意、上限 bytes)
@@ -189,6 +193,9 @@ optional env:
 - `RELAY_TRIGGER_REF_PREFIXES` (CSV, default: `refs/relay/incoming/`)
 - `RELAY_CONFIG_JSON` (JSON override)
 
+`RELAY_REPO_ID` を指定しない場合は `git remote.origin.url` と直近 commit を参照して、 peer cache
+sync の対象 repo を自動推定します（`deno task dev` は `git` 実行権限付き）。
+
 互換モード（従来の unsigned publish 許可）:
 
 ```bash

deno.json

@@ -9,7 +9,7 @@
     "test:unit": "deno task test:layer unit",
     "test:integration": "deno task test:layer integration",
     "test:e2e": "deno task test:layer e2e",
-    "dev": "deno run --allow-net --allow-env --watch src/deno_main.ts",
+    "dev": "deno run --allow-net --allow-env --allow-run=git --watch src/deno_main.ts",
     "mcp": "deno run --allow-run --allow-net --allow-env --allow-read src/mcp_server.ts",
     "fmt": "deno fmt",
     "lint": "deno lint"

src/deno_main.ts

@@ -16,6 +16,12 @@ import {
   safeWriteGitCache,
 } from './git_cache_layer.ts';
 import { createWebhookTriggerDispatcher } from './trigger_dispatcher.ts';
+import {
+  decideRepositoryCompatibility,
+  parseRepositoryFromDiscoveryBody,
+  type RelayRepositoryIdentity,
+  resolveLocalRepositoryIdentity,
+} from './repository_affinity.ts';
 
 function parsePositiveInt(raw: string | undefined, fallback: number): number {
   const value = Number.parseInt(raw ?? '', 10);
@@ -25,6 +31,17 @@ function parsePositiveInt(raw: string | undefined, fallback: number): number {
   return Math.trunc(value);
 }
 
+function parseCsvList(raw: string | undefined): string[] {
+  if (typeof raw !== 'string' || raw.trim().length === 0) return [];
+  const dedupe = new Set<string>();
+  for (const part of raw.split(',')) {
+    const value = part.trim();
+    if (value.length === 0) continue;
+    dedupe.add(value);
+  }
+  return [...dedupe];
+}
+
 const SESSION_ID_PATTERN = /^[A-Za-z0-9]{6,16}$/;
 const NAMED_SESSION_PATTERN = /^[A-Za-z0-9][A-Za-z0-9._-]{0,38}\/[A-Za-z0-9][A-Za-z0-9._-]{0,63}$/;
 
@@ -46,6 +63,16 @@ function generateSessionId(): string {
 const host = Deno.env.get('HOST') ?? '127.0.0.1';
 const port = parsePositiveInt(Deno.env.get('PORT') ?? undefined, 8788);
 const runtimeConfig = parseRelayRuntimeConfigFromEnv((key) => Deno.env.get(key) ?? undefined);
+const repoFfCommitWindow = parsePositiveInt(
+  Deno.env.get('RELAY_PEER_REPO_FF_COMMIT_WINDOW') ?? undefined,
+  30,
+);
+const localRepositoryIdentity = await resolveLocalRepositoryIdentity({
+  explicitRepoId: Deno.env.get('RELAY_REPO_ID') ?? undefined,
+  explicitOriginUrl: Deno.env.get('RELAY_REPO_ORIGIN_URL') ?? undefined,
+  explicitRecentCommits: parseCsvList(Deno.env.get('RELAY_REPO_RECENT_COMMITS') ?? undefined),
+  commitWindow: repoFfCommitWindow,
+});
 const relayCacheStore = createMemoryCacheStore({
   ttlSec: runtimeConfig.cache.ttlSec,
   maxBytes: runtimeConfig.cache.maxBytes,
@@ -55,6 +82,10 @@ const service = createMemoryRelayService({
   peerRelayUrls: runtimeConfig.peers.urls.length > 0
     ? runtimeConfig.peers.urls
     : runtimeConfig.relay.peerRelayUrls,
+  repositoryId: localRepositoryIdentity.repoId ?? undefined,
+  repositoryOwner: localRepositoryIdentity.owner ?? undefined,
+  repositoryName: localRepositoryIdentity.name ?? undefined,
+  repositoryRecentCommits: localRepositoryIdentity.recentCommits,
   cacheStore: relayCacheStore,
   githubWebhookSecret: runtimeConfig.github.webhookSecret ?? undefined,
 });
@@ -117,6 +148,11 @@ function parseCacheExchangePullBody(body: Record<string, unknown>): {
   return { entries, nextCursor };
 }
 
+function parsePeerNodeIdFromDiscoveryBody(body: Record<string, unknown>): string | null {
+  const raw = typeof body.node_id === 'string' ? body.node_id.trim() : '';
+  return raw.length > 0 ? raw : null;
+}
+
 function nowEpochSec(): number {
   return Math.floor(Date.now() / 1000);
 }
@@ -173,10 +209,84 @@ async function startCacheSyncWorker(): Promise<void> {
     return;
   }
 
+  const repositoryScopedSyncEnabled = localRepositoryIdentity.name !== null;
+  if (!repositoryScopedSyncEnabled) {
+    console.warn(
+      '[bit-relay] repository affinity disabled: set RELAY_REPO_ID or enable git origin detection',
+    );
+  }
+
+  const discoveryCacheTtlMs = Math.max(5, runtimeConfig.peers.syncIntervalSec) * 1000;
+  const loggedRepositoryMismatchPeers = new Set<string>();
+  const peerDiscoveryCache = new Map<
+    string,
+    { expiresAt: number; nodeId: string | null; repository: RelayRepositoryIdentity | null }
+  >();
+
+  async function fetchPeerDiscovery(
+    peer: string,
+  ): Promise<{ nodeId: string | null; repository: RelayRepositoryIdentity | null }> {
+    const url = new URL('/api/v1/cache/exchange/discovery', peer);
+    const headers = new Headers();
+    if (peerSyncAuthToken.length > 0) {
+      headers.set('authorization', `Bearer ${peerSyncAuthToken}`);
+    }
+    const response = await fetch(url.toString(), {
+      method: 'GET',
+      headers,
+    });
+    if (response.status !== 200) {
+      const text = await response.text();
+      throw new Error(
+        `peer discovery failed: peer=${peer}, status=${response.status}, body=${text}`,
+      );
+    }
+    const body = await response.json() as Record<string, unknown>;
+    return {
+      nodeId: parsePeerNodeIdFromDiscoveryBody(body),
+      repository: parseRepositoryFromDiscoveryBody(body, repoFfCommitWindow),
+    };
+  }
+
+  async function resolvePeerDiscovery(
+    peer: string,
+  ): Promise<{ nodeId: string | null; repository: RelayRepositoryIdentity | null }> {
+    const cached = peerDiscoveryCache.get(peer);
+    if (cached && Date.now() < cached.expiresAt) {
+      return { nodeId: cached.nodeId, repository: cached.repository };
+    }
+    const discovered = await fetchPeerDiscovery(peer);
+    peerDiscoveryCache.set(peer, {
+      expiresAt: Date.now() + discoveryCacheTtlMs,
+      nodeId: discovered.nodeId,
+      repository: discovered.repository,
+    });
+    return discovered;
+  }
+
   const worker = createCacheSyncWorker({
     peers,
     limit: 200,
     async pullFromPeer({ peer, after, limit }) {
+      if (repositoryScopedSyncEnabled) {
+        const peerDiscovery = await resolvePeerDiscovery(peer);
+        const decision = decideRepositoryCompatibility(
+          localRepositoryIdentity,
+          peerDiscovery.repository,
+        );
+        if (!decision.compatible) {
+          if (!loggedRepositoryMismatchPeers.has(peer)) {
+            const peerNode = peerDiscovery.nodeId ?? peer;
+            console.warn(
+              `[bit-relay] cache-sync skipped peer=${peerNode} reason=${decision.reason}`,
+            );
+            loggedRepositoryMismatchPeers.add(peer);
+          }
+          return { entries: [], nextCursor: after };
+        }
+        loggedRepositoryMismatchPeers.delete(peer);
+      }
+
       const url = new URL('/api/v1/cache/exchange/pull', peer);
       url.searchParams.set('after', String(after));
       url.searchParams.set('limit', String(limit));

src/memory_handler.ts

@@ -197,6 +197,10 @@ export interface MemoryRelayOptions {
   wsIdleTimeoutMs?: number;
   relayNodeId?: string;
   peerRelayUrls?: string[];
+  repositoryId?: string;
+  repositoryOwner?: string;
+  repositoryName?: string;
+  repositoryRecentCommits?: string[];
   cacheExchangeMaxHops?: number;
   cacheExchangeMaxRecords?: number;
   cacheStore?: CacheStore;
@@ -238,6 +242,7 @@ const MAX_GITHUB_WEBHOOK_DLQ_ENTRIES = 10_000;
 const GITHUB_WEBHOOK_RETRY_BASE_SEC = 30;
 const INCOMING_TRIGGER_REF_PREFIX = 'refs/relay/incoming/';
 const ROOM_NAME_PATTERN = /^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$/;
+const COMMIT_HASH_PATTERN = /^[0-9a-f]{7,40}$/i;
 const TOPIC_PATTERN = /^[a-z][a-z0-9._-]{0,63}$/;
 const PR_ID_PATTERN = /^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$/;
 const NODE_ID_PATTERN = /^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$/;
@@ -330,6 +335,66 @@ function normalizePeerRelayUrls(raw: string[] | undefined): string[] {
   return [...dedupe];
 }
 
+interface NormalizedRepositoryMetadata {
+  repoId: string | null;
+  owner: string | null;
+  name: string | null;
+  recentCommits: string[];
+}
+
+function normalizeRepositoryToken(raw: string | undefined): string | null {
+  const value = (raw ?? '').trim().toLowerCase();
+  if (value.length === 0) return null;
+  if (!/^[a-z0-9._-]+$/.test(value)) return null;
+  return value;
+}
+
+function normalizeRepositoryCommits(raw: string[] | undefined): string[] {
+  if (!Array.isArray(raw)) return [];
+  const dedupe = new Set<string>();
+  for (const value of raw) {
+    if (typeof value !== 'string') continue;
+    const normalized = value.trim().toLowerCase();
+    if (!COMMIT_HASH_PATTERN.test(normalized)) continue;
+    dedupe.add(normalized);
+    if (dedupe.size >= 30) break;
+  }
+  return [...dedupe];
+}
+
+function parseOwnerNameFromRepositoryId(
+  repoId: string,
+): { owner: string | null; name: string | null } {
+  const trimmed = repoId.trim().replace(/^github:/i, '');
+  const cleaned = trimmed.endsWith('.git') ? trimmed.slice(0, -4) : trimmed;
+  const segments = cleaned.split('/').map((entry) => entry.trim()).filter((entry) =>
+    entry.length > 0
+  );
+  if (segments.length === 0) {
+    return { owner: null, name: null };
+  }
+  const name = normalizeRepositoryToken(segments[segments.length - 1]) ?? null;
+  const owner = segments.length >= 2
+    ? normalizeRepositoryToken(segments[segments.length - 2])
+    : null;
+  return { owner, name };
+}
+
+function normalizeRepositoryMetadata(options: MemoryRelayOptions): NormalizedRepositoryMetadata {
+  const repoIdRaw = (options.repositoryId ?? '').trim();
+  const repoId = repoIdRaw.length > 0 ? repoIdRaw.replace(/^github:/i, '').toLowerCase() : null;
+  const fromRepoId = repoId ? parseOwnerNameFromRepositoryId(repoId) : { owner: null, name: null };
+  const owner = normalizeRepositoryToken(options.repositoryOwner) ?? fromRepoId.owner;
+  const name = normalizeRepositoryToken(options.repositoryName) ?? fromRepoId.name;
+  const recentCommits = normalizeRepositoryCommits(options.repositoryRecentCommits);
+  return {
+    repoId,
+    owner,
+    name,
+    recentCommits,
+  };
+}
+
 function isValidRoomName(room: string): boolean {
   return ROOM_NAME_PATTERN.test(room);
 }
@@ -1088,6 +1153,7 @@ export function createMemoryRelayService(options: MemoryRelayOptions = {}): Memo
   );
   const relayNodeId = normalizeRelayNodeId(options.relayNodeId);
   const peerRelayUrls = normalizePeerRelayUrls(options.peerRelayUrls);
+  const repositoryMetadata = normalizeRepositoryMetadata(options);
   const cacheExchangeMaxHops = Math.max(
     1,
     Math.trunc(options.cacheExchangeMaxHops ?? DEFAULT_CACHE_EXCHANGE_MAX_HOPS),
@@ -1330,6 +1396,12 @@ export function createMemoryRelayService(options: MemoryRelayOptions = {}): Memo
       node_id: relayNodeId,
       peers: peerRelayUrls,
       max_hops: cacheExchangeMaxHops,
+      repository: {
+        repo_id: repositoryMetadata.repoId,
+        owner: repositoryMetadata.owner,
+        name: repositoryMetadata.name,
+        recent_commits: repositoryMetadata.recentCommits,
+      },
     }, { status: 200 });
   }