| … | … | @@ -24,7 +24,7 @@ import { RepoStore } from 'atproto-store/lib/repo-store.js' |
| 24 | 24 | import { FirehoseSequencer } from 'atproto-store/lib/sequencer.js' |
| 25 | 25 | import { decodeCommitFrame, syncFrame } from 'atproto-store/lib/firehose.js' |
| 26 | 26 | import { getRepoFromCommit, getRecord as rvGetRecord, listRecords as rvListRecords } from 'atproto-store/lib/read-view.js' |
| 27 | | import { readCar, loadCommit, decodeBlock } from 'atproto-repo/lib/repo.js' |
| 27 | import { readCar, loadCommit, decodeBlock, listRecords as carListRecords } from 'atproto-repo/lib/repo.js' |
| 28 | 28 | import { verifyCommit, commitCid } from 'atproto-repo/lib/commit.js' |
| 29 | 29 | |
| 30 | 30 | const ANNOUNCE_PROTOCOL = 'atproto/did-announce/v1' |
| … | … | @@ -77,6 +77,25 @@ export class DidPeer { |
| 77 | 77 | // them so we can serve + announce. Idempotent. |
| 78 | 78 | async provide ({ repoKey, firehoseKey } = {}) { return this._openCores(repoKey, firehoseKey) } |
| 79 | 79 | |
| 80 | // Re-open a repo core for a front that went back to PROJECTING after a stateless spell. The router |
| 81 | // drops the replica when a peer announces repoKey:null; without this the drop is ONE-WAY, so a peer |
| 82 | // that resumes projecting stays repo-less for the life of the process — permanently on the slower |
| 83 | // derive path. Observed: rolling projectRepoCore back to true did NOT heal the cached front. |
| 84 | async adoptRepoKey (repoKey) { |
| 85 | if (this._noRepo || this.repo || !repoKey) return false |
| 86 | // CRITICAL: during the INITIAL _openCores, this.repo is null while the open is in flight. Without |
| 87 | // this guard a second announce arriving mid-open races it and opens a COMPETING repo core for the |
| 88 | // same front — which deterministically broke the co-owned freshest-by-rev test (the ahead co-owner |
| 89 | // was never fronted). Only adopt for a front that is already established (seq open, nothing opening). |
| 90 | if (this._opening || !this.seq) return false |
| 91 | this._repoKey = toKey(repoKey) |
| 92 | this.repo = await new RepoStore(this.corestore, { key: this._repoKey }).ready() |
| 93 | try { this.repo.core.download({ start: 0, end: -1 }) } catch {} |
| 94 | this._derived = null // the derive cache is now redundant; the core is authoritative |
| 95 | this._derivedAt = null |
| 96 | return true |
| 97 | } |
| 98 | |
| 80 | 99 | async _openCores (repoKey, firehoseKey) { |
| 81 | 100 | if (this.seq) return // firehose is opened in BOTH modes; the idempotency guard (repo may stay null under noRepo) |
| 82 | 101 | if (this._opening) return this._opening |
| … | … | @@ -231,16 +250,49 @@ export class DidPeer { |
| 231 | 250 | return { carBytes, commitCid: commitCid(signed), dataCid: signed.data } |
| 232 | 251 | } |
| 233 | 252 | |
| 253 | // The record set for a peer with NO repo core, derived from the CAR getRepo() already fetches and |
| 254 | // VERIFIES. Cached against the head rev, so N reads at one head cost ONE derive and the cache |
| 255 | // invalidates the moment a new commit lands — no staleness, no per-read CAR fetch. |
| 256 | // |
| 257 | // Without this, getRecord/listRecords/collections returned null/[] whenever this.repo was null. That |
| 258 | // was tolerable when the only repo-less mode was noRepo (favela, which reads records from the shared |
| 259 | // store instead), but it silently broke com.atproto.repo.getRecord the moment a peer went STATELESS — |
| 260 | // observed in production as RecordNotFound for a profile that was present and correctly served by |
| 261 | // getRepo. The trust story is unchanged: these read out of the same verified CAR. |
| 262 | async _derivedRecords () { |
| 263 | if (!this.head) return [] |
| 264 | // Derive ONLY when it can actually work. A front with no repo core and no RPC (one still mid-open, |
| 265 | // or a plain seeder) previously returned an empty list from these readers; making it THROW instead |
| 266 | // changed observable behaviour for every such caller — it broke the co-owned freshest-by-rev test, |
| 267 | // where a poller reading a not-yet-open front started erroring instead of reporting "nothing yet". |
| 268 | if (!this.rpc) return [] |
| 269 | if (this._derivedAt === this.head.rev && this._derived) return this._derived |
| 270 | const { carBytes } = await this.getRepo() |
| 271 | const car = await readCar(carBytes) |
| 272 | // listRecords already yields the record CID (the MST value), so nothing is recomputed and no |
| 273 | // dag-cbor dependency is pulled in here — the CAR is the single source for key, cid and value. |
| 274 | this._derived = carListRecords(car).map((r) => ({ key: r.key, value: r.value, cid: r.cid.toString() })) |
| 275 | this._derivedAt = this.head.rev |
| 276 | return this._derived |
| 277 | } |
| 278 | |
| 234 | 279 | // com.atproto.repo.getRecord |
| 235 | | getRecord (collection, rkey) { |
| 236 | | if (!this.repo) return null |
| 237 | | return rvGetRecord(this.repo, collection, rkey, { did: this.did }) |
| 280 | async getRecord (collection, rkey) { |
| 281 | if (this.repo) return rvGetRecord(this.repo, collection, rkey, { did: this.did }) |
| 282 | if (this._noRepo) return null // records live in the consumer's own store, by design |
| 283 | const key = collection + '/' + rkey |
| 284 | const hit = (await this._derivedRecords()).find((r) => r.key === key) |
| 285 | return hit ? { uri: `at://${this.did}/${key}`, cid: hit.cid, value: hit.value } : null |
| 238 | 286 | } |
| 239 | 287 | |
| 240 | 288 | // com.atproto.repo.listRecords |
| 241 | | listRecords (collection) { |
| 242 | | if (!this.repo) return [] |
| 243 | | return rvListRecords(this.repo, collection, { did: this.did }) |
| 289 | async listRecords (collection) { |
| 290 | if (this.repo) return rvListRecords(this.repo, collection, { did: this.did }) |
| 291 | if (this._noRepo) return [] |
| 292 | const prefix = collection + '/' |
| 293 | return (await this._derivedRecords()) |
| 294 | .filter((r) => r.key.startsWith(prefix)) |
| 295 | .map((r) => ({ uri: `at://${this.did}/${r.key}`, cid: r.cid, value: r.value })) |
| 244 | 296 | } |
| 245 | 297 | |
| 246 | 298 | // com.atproto.sync.getLatestCommit -> { cid, rev } from the verified head (keyless). |
| … | … | @@ -315,9 +367,9 @@ export class DidPeer { |
| 315 | 367 | |
| 316 | 368 | // Distinct collection NSIDs present in the replicated repo (keys are "<collection>/<rkey>"). |
| 317 | 369 | async collections () { |
| 318 | | if (!this.repo) return [] |
| 319 | 370 | const set = new Set() |
| 320 | | for (const r of await this.repo.records()) { |
| 371 | const rows = this.repo ? await this.repo.records() : (this._noRepo ? [] : await this._derivedRecords()) |
| 372 | for (const r of rows) { |
| 321 | 373 | const i = r.key.indexOf('/') |
| 322 | 374 | if (i > 0) set.add(r.key.slice(0, i)) |
| 323 | 375 | } |
| … | … | @@ -501,6 +553,10 @@ export class DidPeerRouter { |
| 501 | 553 | existing.repo = null |
| 502 | 554 | existing._repoKey = null |
| 503 | 555 | try { await stale.close() } catch {} |
| 556 | } else if (a.repoKey && !existing.repo) { |
| 557 | // ...and the reverse: the peer resumed projecting, so take the core back. Without this the drop |
| 558 | // above is one-way and a rolled-back flip never heals. |
| 559 | existing.adoptRepoKey(a.repoKey).catch(() => {}) |
| 504 | 560 | } |
| 505 | 561 | if (existing.seq) existing.seq.core.update().catch(() => {}) |
| 506 | 562 | this._wake(did) |